From 26ac61a00ad2868c577b47aaaed3adf55c841be3 Mon Sep 17 00:00:00 2001 From: overtrue Date: Mon, 7 Sep 2026 23:27:12 +0800 Subject: [PATCH] test(ecstore): cover user source ownership during target rename --- crates/ecstore/src/store/peer.rs | 141 +++++++++++++++++++++++++++++++ 1 file changed, 141 insertions(+) diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 858b14be7..2d74a6535 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -765,6 +765,147 @@ mod tests { } } + #[cfg(not(windows))] + #[tokio::test] + async fn target_user_source_to_internal_destination_retains_owner_after_cancellation() { + use crate::disk::os::prepared_publication_test_hooks as hooks; + use futures::FutureExt; + use std::time::Duration; + + let ctx = Arc::new(InstanceContext::new()); + let root = tempfile::tempdir().expect("source-volume root"); + let disk = target_disk(&ctx, root.path(), Uuid::new_v4()).await; + let store = super::super::tests::build_store_with_ctx(ctx.clone()); + let version = Uuid::new_v4(); + let fi = target_file_info("object", version, b"user-source-inline-body"); + fi.validate_for_metadata_read().expect("valid real inline metadata"); + let source_before = seed_target(&disk, "photos", "object", fi.clone()).await; + assert!(!source_before.is_empty()); + tokio::fs::create_dir_all(root.path().join(crate::disk::RUSTFS_META_TMP_BUCKET)) + .await + .expect("internal staging volume"); + let destination = disk + .get_object_path_for_io_if_local(crate::disk::RUSTFS_META_TMP_BUCKET, "object") + .expect("local disk") + .expect("actual destination object key"); + let destination_metadata = destination.join(crate::disk::STORAGE_FORMAT_FILE); + assert!(!destination_metadata.exists()); + assert!(!ctx.namespace_commits_pending()); + let generation = ctx.namespace_commit_generation(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let _hook = hooks::install(&destination_metadata, move || { + let _ = entered_tx.send(()); + let _ = release_rx.recv(); + }); + let disk_ref = disk.endpoint().to_string(); + let rename_fi = fi.clone(); + let mut rename = tokio::spawn(async move { + store + .rename_local_data( + &disk_ref, + ("photos", "object"), + &rename_fi, + (crate::disk::RUSTFS_META_TMP_BUCKET, "object"), + None, + ) + .await + }); + let mut entered = false; + let mut joined = false; + let observations = std::panic::AssertUnwindSafe(async { + tokio::time::timeout(Duration::from_secs(10), async { + tokio::select! { + result = &mut rename => { + joined = true; + panic!("rename completed before prepared publication: {result:?}"); + } + result = entered_rx => { + result.expect("real prepared rename must enter"); + entered = true; + } + } + }) + .await + .expect("bounded physical entry"); + let at_entry = (ctx.namespace_commits_pending(), ctx.namespace_commit_generation()); + assert!(!rename.is_finished(), "caller must still await the paused physical rename"); + rename.abort(); + let cancelled = tokio::time::timeout(Duration::from_secs(5), &mut rename) + .await + .expect("caller cancellation must finish while physical publication is paused"); + joined = true; + let after_cancel = (ctx.namespace_commits_pending(), ctx.namespace_commit_generation()); + (at_entry, after_cancel, cancelled) + }) + .catch_unwind() + .await; + + // Release on every observation failure. Pending alone is not a drain + // oracle: the implementation under test can fail to create the owner. + drop(release_tx); + if !joined { + rename.abort(); + joined = tokio::time::timeout(Duration::from_secs(5), &mut rename).await.is_ok(); + } + let physical_drained = tokio::time::timeout(Duration::from_secs(10), hooks::drain_namespace_key(&destination)).await; + let owner_drained = tokio::time::timeout(Duration::from_secs(10), async { + while ctx.namespace_commits_pending() { + tokio::task::yield_now().await; + } + }) + .await; + if !entered || !joined || physical_drained.is_err() || owner_drained.is_err() { + // Without proven physical entry/drain, keep the root instead of + // deleting files that a detached local executor may still use. + let retained = root.keep(); + eprintln!("source-volume cleanup incomplete: entered={entered}, joined={joined}, retained={retained:?}"); + if let Err(panic) = observations { + std::panic::resume_unwind(panic); + } + panic!("source-volume physical cleanup did not finish: retained={retained:?}"); + } + let (at_entry, after_cancel, cancelled) = match observations { + Ok(observations) => observations, + Err(panic) => std::panic::resume_unwind(panic), + }; + let latest = tokio::time::timeout( + Duration::from_secs(5), + disk.read_version( + crate::disk::RUSTFS_META_TMP_BUCKET, + crate::disk::RUSTFS_META_TMP_BUCKET, + "object", + "", + &crate::disk::ReadOptions { + read_data: true, + ..Default::default() + }, + ), + ) + .await; + let source = + tokio::time::timeout(Duration::from_secs(5), tokio::fs::read(root.path().join("photos/object/xl.meta"))).await; + let after_drain = (ctx.namespace_commits_pending(), ctx.namespace_commit_generation()); + assert!(cancelled.expect_err("caller must return cancellation").is_cancelled()); + let latest = latest + .expect("latest read must finish") + .expect("late physical commit must be readable"); + assert_eq!(latest.data, fi.data); + assert_eq!(latest.version_id, Some(version)); + assert_eq!( + source + .expect("source observation must finish") + .expect_err("user source metadata must have moved") + .kind(), + std::io::ErrorKind::NotFound + ); + assert_eq!( + (at_entry, after_cancel, after_drain), + ((true, generation + 1), (true, generation + 1), (false, generation + 2)), + "a real user source mutation must remain counted through its cancelled caller and physical drain" + ); + } + #[cfg(not(windows))] #[tokio::test] async fn target_rename_cancellation_retains_real_namespace_and_scanner_owners() {