diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index e5eccba32..b98b294ca 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -317,6 +317,22 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper { dst_path: &str, external_guard: Option>, ) -> Result { + self.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, external_guard) + .await + .result + } +} + +impl LocalDiskWrapper { + pub(in crate::disk) async fn rename_data_observed( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + external_guard: Option>, + ) -> super::RenameDataObservation { let operation = self.clone(); let src_volume = src_volume.to_owned(); let src_path = src_path.to_owned(); @@ -333,22 +349,35 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper { } else { get_max_timeout_duration() }; - run_owned_mutation(external_guard, move || async move { - operation + let observed = run_owned_mutation(external_guard, move || async move { + let mut preflight_rejection = None; + let result = operation .track_disk_health_mutation( "rename_data", DiskMetricMutation::Write, || async { - operation - .disk - .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path) - .await + // Preserve the former DiskAPI future's single boxing boundary. + let observed = + Box::pin( + operation + .disk + .rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path), + ) + .await; + preflight_rejection = observed.preflight_rejection; + observed.result }, timeout_duration, ) - .await + .await; + // Health tracking must observe the real disk error, not an Ok tuple. + Ok(super::RenameDataObservation { + result, + preflight_rejection, + }) }) - .await + .await; + observed.unwrap_or_else(|error| super::RenameDataObservation::unknown(Err(error))) } } @@ -2588,6 +2617,46 @@ mod tests { assert_eq!(wrapper.metrics_snapshot().api_calls.get("unknown"), Some(&1)); } + #[tokio::test] + async fn rename_preflight_evidence_preserves_health_errors_and_owned_reply() { + for source_exists in [false, true] { + for guarded in [false, true] { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")) + .expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + if source_exists { + disk.make_volume("source").await.expect("source volume should exist"); + } + let wrapper = LocalDiskWrapper::new(disk, false); + let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let external_guard = guarded.then(|| Arc::new(DropProbe(Arc::clone(&drops))) as Arc); + let mut file_info = FileInfo::new("object", 1, 0); + file_info.mod_time = Some(::time::OffsetDateTime::now_utc()); + file_info.erasure.index = 1; + let observed = wrapper + .rename_data_observed("source", "object", &file_info, "missing-destination", "object", external_guard) + .await; + assert!(observed.rejected_before_publication(), "normal access rejection must carry proof"); + assert!(matches!(observed.result, Err(DiskError::VolumeNotFound))); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1)); + assert_eq!(snapshot.total_writes, 0, "health tracking must not observe the rejection as Ok"); + assert_eq!(drops.load(Ordering::SeqCst), usize::from(guarded)); + + wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + let observed = wrapper + .rename_data_observed("source", "object", &file_info, "missing-destination", "object", None) + .await; + assert!(!observed.rejected_before_publication(), "wrapper errors carry no local preflight proof"); + assert!(matches!(observed.result, Err(DiskError::FaultyDisk))); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.total_errors_availability, 1); + assert_eq!(snapshot.total_writes, 0); + } + } + } + #[tokio::test] async fn local_disk_health_wrapper_counts_returned_availability_errors() { let dir = tempfile::tempdir().expect("temp dir should be created"); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 774450201..9cd683578 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(in crate::disk) use self::commit::LocalRenamePreflightRejection; #[cfg(test)] use self::commit::lock_rename_commit_directories; @@ -8866,7 +8867,6 @@ impl DiskAPI for LocalDisk { Ok(()) } - #[tracing::instrument(level = "trace", skip_all)] async fn rename_data( &self, src_volume: &str, @@ -8875,8 +8875,8 @@ impl DiskAPI for LocalDisk { dst_volume: &str, dst_path: &str, ) -> Result { - crate::hp_guard!("LocalDisk::rename_data"); - self.rename_data_commit(src_volume, src_path, fi, dst_volume, dst_path).await + self.rename_data_inner(src_volume, src_path, fi, dst_volume, dst_path, &mut None) + .await } #[tracing::instrument(level = "trace", skip_all)] @@ -12971,6 +12971,196 @@ mod test { ); } + #[tokio::test] + async fn observed_rename_timeout_has_no_preflight_proof_and_retains_namespace_lease() { + use crate::disk::disk_store::LocalDiskWrapper; + use futures::FutureExt; + use std::sync::mpsc; + + temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let bucket = "observed-timeout-bucket"; + let object = "prefix/object"; + let tmp_object = "observed-timeout-stage"; + let data_dir = Uuid::new_v4(); + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + let staged_part = disk + .get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}/part.1")) + .expect("staged part path should resolve"); + fs::create_dir_all(staged_part.parent().expect("staged part should have a parent")) + .await + .expect("staged data directory should be created"); + fs::write(&staged_part, b"new-payload") + .await + .expect("staged part should be written"); + let staged_metadata = disk + .get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{STORAGE_FORMAT_FILE}")) + .expect("staged metadata path should resolve"); + let destination = disk.io_get_object_path(bucket, object).expect("destination should resolve"); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + set_owned_file_write_before_open(&staged_metadata, move || { + entered_tx.send(()).expect("signal staged writer entry"); + // Dropping the sender also unblocks the syscall if the test fails. + let _ = release_rx.recv(); + }); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + let operation = wrapper.clone(); + let fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None); + let rename = tokio::spawn(async move { + operation + .rename_data_observed(RUSTFS_META_TMP_BUCKET, tmp_object, &fi, bucket, object, None) + .await + }); + tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(10))) + .await + .expect("staged writer waiter should run") + .expect("rename must enter the real staged metadata write"); + + // Advance only after the blocking syscall owns its lease and the wrapper's timer exists. + tokio::time::pause(); + tokio::time::advance(Duration::from_secs(61)).await; + tokio::time::resume(); + let observed = tokio::time::timeout(Duration::from_secs(5), rename) + .await + .expect("wrapper timeout must not wait for the blocked syscall") + .expect("the wrapper waiter must not panic"); + assert!(!observed.rejected_before_publication(), "a timeout must carry no local preflight proof"); + assert!(matches!(observed.result, Err(DiskError::Timeout))); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1)); + assert_eq!(snapshot.total_errors_timeout, 1); + assert_eq!(snapshot.total_writes, 0); + assert_eq!(snapshot.total_waiting, 0); + let volume_lock = os::disk_volume_mutation_lock(&disk.root, bucket); + assert!( + Arc::clone(&volume_lock).try_write_owned().is_err(), + "the blocked syscall must retain its volume guard" + ); + assert!( + os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination) + .now_or_never() + .is_none(), + "a same-object mutation must still wait for the blocked syscall" + ); + assert_eq!(fs::read(&staged_part).await.expect("staged data must remain"), b"new-payload"); + assert!(!destination.join(STORAGE_FORMAT_FILE).exists()); + + release_tx.send(()).expect("release timed-out staged writer"); + let lease = tokio::time::timeout( + Duration::from_secs(5), + os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination), + ) + .await + .expect("the namespace lease must be released when the syscall drains"); + drop(lease); + let _exclusive = tokio::time::timeout(Duration::from_secs(5), volume_lock.write_owned()) + .await + .expect("the volume guard must be released when the syscall drains"); + assert!( + !destination.join(STORAGE_FORMAT_FILE).exists(), + "timed-out waiter must not publish metadata later" + ); + }) + .await; + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn observed_rename_owned_task_panic_has_no_preflight_proof_and_releases_guard() { + use crate::disk::disk_store::LocalDiskWrapper; + use std::sync::mpsc; + + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let bucket = "observed-panic-bucket"; + let object = "prefix/object"; + let tmp_object = "observed-panic-stage"; + let data_dir = Uuid::new_v4(); + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + let staged_part = disk + .get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}/part.1")) + .expect("staged part path should resolve"); + fs::create_dir_all(staged_part.parent().expect("staged part should have a parent")) + .await + .expect("staged data directory should be created"); + fs::write(&staged_part, b"new-payload") + .await + .expect("staged part should be written"); + let destination = disk.io_get_object_path(bucket, object).expect("destination should resolve"); + let published_part = destination.join(data_dir.to_string()).join("part.1"); + let (entered_tx, entered_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + set_rename_data_after_first_publication(&disk.root, bucket, object, move || { + entered_tx.send(()).expect("signal data publication"); + let _ = release_rx.recv(); + // This hook runs in the owned async mutation, outside spawn_blocking. + panic!("injected observed rename owner panic after publication"); + }); + let external_guard = Arc::new(()); + let guard_probe = Arc::downgrade(&external_guard); + let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false); + let operation = wrapper.clone(); + let fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None); + let rename = tokio::spawn(async move { + operation + .rename_data_observed(RUSTFS_META_TMP_BUCKET, tmp_object, &fi, bucket, object, Some(external_guard)) + .await + }); + tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(10))) + .await + .expect("publication waiter should run") + .expect("rename must publish data before the injected owner panic"); + assert!(guard_probe.upgrade().is_some(), "the owned task must retain the publication guard"); + assert_eq!(fs::read(&published_part).await.expect("new data must be published"), b"new-payload"); + assert!(!staged_part.exists(), "the real data rename must have consumed staging"); + assert!(!destination.join(STORAGE_FORMAT_FILE).exists()); + let volume_lock = os::disk_volume_mutation_lock(&disk.root, bucket); + assert!( + Arc::clone(&volume_lock).try_write_owned().is_err(), + "the mutation must retain its volume guard" + ); + + release_tx.send(()).expect("release mutation owner into the injected panic"); + let observed = tokio::time::timeout(Duration::from_secs(5), rename) + .await + .expect("owned task panic must reach the wrapper") + .expect("the wrapper must convert the inner task panic into an error"); + assert!( + !observed.rejected_before_publication(), + "a join failure must carry no local preflight proof" + ); + assert!(matches!(observed.result, Err(DiskError::Io(error)) if error.to_string() == "owned mutation task failed")); + assert!( + guard_probe.upgrade().is_none(), + "the guard must be released after the mutation owner unwinds" + ); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1)); + assert_eq!(snapshot.total_writes, 0); + assert_eq!(snapshot.total_waiting, 0); + let lease = tokio::time::timeout( + Duration::from_secs(5), + os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination), + ) + .await + .expect("panic must release the namespace lease"); + drop(lease); + let _exclusive = tokio::time::timeout(Duration::from_secs(5), volume_lock.write_owned()) + .await + .expect("panic must release the volume guard"); + assert_eq!( + fs::read(&published_part).await.expect("published recovery data must remain"), + b"new-payload" + ); + assert!(!destination.join(STORAGE_FORMAT_FILE).exists()); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn windows_and_unix_cancelled_staged_metadata_write_serializes_same_object_retry() { use std::sync::{Arc, mpsc}; diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index 12663bc5f..9d51bc1cd 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Single-disk object rename publication and rollback. The caller retains the -//! DiskAPI instrumentation; mutation leases and commit guards follow the syscall. +//! Single-disk object rename publication and rollback. The shared execution core +//! retains instrumentation, mutation leases, and commit guards through the syscall. #[cfg(all(test, windows))] use super::run_destination_commit_directory_preparation; @@ -226,15 +226,23 @@ async fn restore_published_data_source( restore_renamed_data_source(src_volume_dir, src_data_path, dst_data_path, publication_root, mutation_lease).await } +/// Proof produced only when the local rename returns at an existing access +/// preflight, before metadata, backups, or object data can be published. +#[derive(Debug)] +pub(in crate::disk) struct LocalRenamePreflightRejection(()); + impl LocalDisk { - pub(super) async fn rename_data_commit( + #[tracing::instrument(name = "rename_data", target = "rustfs_ecstore::disk::local", level = "trace", skip_all)] + pub(super) async fn rename_data_inner( &self, src_volume: &str, src_path: &str, fi: FileInfo, dst_volume: &str, dst_path: &str, + preflight_rejection: &mut Option, ) -> Result { + crate::hp_guard!("LocalDisk::rename_data"); let mut fi = fi; // A non-force DeleteBucket must not remove a directory while a local // object commit is publishing into it. The peer's empty scan remains @@ -294,6 +302,7 @@ impl LocalDisk { error = %e, "Disk local access check failed" ); + *preflight_rejection = Some(LocalRenamePreflightRejection(())); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -311,6 +320,7 @@ impl LocalDisk { error = %e, "Disk local access check failed" ); + *preflight_rejection = Some(LocalRenamePreflightRejection(())); return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } @@ -1202,4 +1212,22 @@ impl LocalDisk { }) } } + + pub(in crate::disk) async fn rename_data_observed( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> super::super::RenameDataObservation { + let mut preflight_rejection = None; + let result = self + .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut preflight_rejection) + .await; + super::super::RenameDataObservation { + result, + preflight_rejection, + } + } } diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 7801274ef..c2f2c52b4 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -75,6 +75,25 @@ use time::OffsetDateTime; use tokio::io::{AsyncRead, AsyncWrite}; use uuid::Uuid; +/// Local preflight evidence stays outside DiskAPI and the RPC response format. +pub(crate) struct RenameDataObservation { + pub(crate) result: Result, + preflight_rejection: Option, +} + +impl RenameDataObservation { + fn unknown(result: Result) -> Self { + Self { + result, + preflight_rejection: None, + } + } + + pub(crate) fn rejected_before_publication(&self) -> bool { + self.result.is_err() && self.preflight_rejection.is_some() + } +} + const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/"; pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token"; @@ -711,6 +730,36 @@ impl Disk { .await } + pub(crate) async fn rename_data_borrowed_with_fence_observed( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + scanner_publication_lease_token: Option, + ) -> RenameDataObservation { + match self { + Disk::Local(local_disk) => { + local_disk + .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None) + .await + } + Disk::Remote(remote_disk) => RenameDataObservation::unknown( + remote_disk + .rename_data_borrowed_with_fence( + src_volume, + src_path, + fi, + dst_volume, + dst_path, + scanner_publication_lease_token, + ) + .await, + ), + } + } + pub(crate) async fn rename_data_borrowed_with_fence( &self, src_volume: &str, diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 676f01cbd..18b04c781 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -3493,9 +3493,17 @@ pub(in crate::set_disk) struct RenameTailOutcome { const EVENT_SET_DISK_RENAME_ROLLBACK: &str = "set_disk_rename_rollback"; +#[derive(Clone, Copy)] +enum RenameDispatchState { + NotDispatched, + RejectedBeforePublication, + MayHavePublished, +} + #[derive(Debug, Clone, PartialEq, Eq)] enum RenameRollbackOutcome { NotAttempted(DiskError), + RejectedBeforePublication(DiskError), Indeterminate(DiskError), Succeeded, Failed(DiskError), @@ -3507,6 +3515,7 @@ impl RenameRollbackOutcome { fn stage(&self) -> &'static str { match self { Self::NotAttempted(_) => "rename_not_dispatched", + Self::RejectedBeforePublication(_) => "rename_rejected_before_publication", Self::Indeterminate(_) => "rename_indeterminate", Self::Succeeded => "undo_succeeded", Self::Failed(_) => "undo_failed", @@ -3592,14 +3601,14 @@ async fn rollback_failed_rename( disks: &[Option], file_infos: Vec, errs: &[Option], - dispatched: &[bool], + dispatch_states: &[RenameDispatchState], rollback_dirs: &[Option], dst: (&str, &str), receipt: Option, ) { let owned_disks = disks.to_vec(); let owned_errs = errs.to_vec(); - let owned_dispatched = dispatched.to_vec(); + let owned_dispatch_states = dispatch_states.to_vec(); let owned_dirs = rollback_dirs.to_vec(); let owned_dst = (dst.0.to_string(), dst.1.to_string()); let coordinator_failure_receipt = receipt.clone(); @@ -3608,7 +3617,7 @@ async fn rollback_failed_rename( let rollback = tokio::spawn(async move { let disks = owned_disks.as_slice(); let errs = owned_errs.as_slice(); - let dispatched = owned_dispatched.as_slice(); + let dispatch_states = owned_dispatch_states.as_slice(); let rollback_dirs = owned_dirs.as_slice(); let dst = (owned_dst.0.as_str(), owned_dst.1.as_str()); let mut file_infos = file_infos; @@ -3619,8 +3628,13 @@ async fn rollback_failed_rename( for (disk_index, disk) in disks.iter().enumerate() { let rollback_dir = rollback_dirs[disk_index]; let outcome = match &errs[disk_index] { - Some(err) if dispatched[disk_index] => RenameRollbackOutcome::Indeterminate(err.clone()), - Some(err) => RenameRollbackOutcome::NotAttempted(err.clone()), + Some(err) => match dispatch_states[disk_index] { + RenameDispatchState::NotDispatched => RenameRollbackOutcome::NotAttempted(err.clone()), + RenameDispatchState::RejectedBeforePublication => { + RenameRollbackOutcome::RejectedBeforePublication(err.clone()) + } + RenameDispatchState::MayHavePublished => RenameRollbackOutcome::Indeterminate(err.clone()), + }, None => RenameRollbackOutcome::Failed(DiskError::DiskNotFound), }; outcomes.push(RenameRollbackDiskOutcome { @@ -4197,7 +4211,7 @@ impl SetDisks { let file_info = file_info.clone(); let successful_rename_completion_rank = successful_rename_completion_rank.clone(); tasks.spawn(async move { - let mut dispatched = false; + let mut dispatch_state = RenameDispatchState::NotDispatched; let result = std::panic::AssertUnwindSafe(async { #[allow(clippy::let_unit_value)] let _fanout_task_guard = Self::rename_fanout_task_guard(&dst_object); @@ -4223,9 +4237,9 @@ impl SetDisks { } let disk_wait_started = rustfs_io_metrics::put_stage_timer(); - dispatched = true; - let result = disk - .rename_data_borrowed_with_fence( + dispatch_state = RenameDispatchState::MayHavePublished; + let observed = disk + .rename_data_borrowed_with_fence_observed( &src_bucket, &src_object, &file_info, @@ -4234,6 +4248,8 @@ impl SetDisks { scanner_publication_lease_token, ) .await; + let rejected_before_publication = observed.rejected_before_publication(); + let result = observed.result; #[cfg(test)] if result.is_ok() { rollback_fault_injection::after_rename(&dst_object, i)?; @@ -4259,11 +4275,14 @@ impl SetDisks { }; rustfs_io_metrics::record_put_rename_disk_wait_completion(position, duration_ms); } + if rejected_before_publication { + dispatch_state = RenameDispatchState::RejectedBeforePublication; + } result }) .catch_unwind() .await; - (i, dispatched, result) + (i, dispatch_state, result) }); } @@ -4274,7 +4293,7 @@ impl SetDisks { let mut results_seen = 0usize; let mut errs = vec![Some(DiskError::DiskNotFound); disk_count]; // Missing task results cannot prove that a disk mutation never ran. - let mut dispatched = vec![true; disk_count]; + let mut dispatch_states = vec![RenameDispatchState::MayHavePublished; disk_count]; let mut disk_versions = vec![None; disk_count]; let mut data_dirs = vec![None; disk_count]; let mut cleanup_data_dirs = vec![None; disk_count]; @@ -4285,8 +4304,8 @@ impl SetDisks { while let Some(joined) = tasks.join_next().await { results_seen += 1; match joined { - Ok((idx, was_dispatched, Ok(Ok(res)))) => { - dispatched[idx] = was_dispatched; + Ok((idx, dispatch_state, Ok(Ok(res)))) => { + dispatch_states[idx] = dispatch_state; data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir); cleanup_data_dirs[idx] = res.cleanup_data_dir; disk_versions[idx] = res.sign; @@ -4294,12 +4313,12 @@ impl SetDisks { errs[idx] = None; success_count += 1; } - Ok((idx, was_dispatched, Ok(Err(err)))) => { - dispatched[idx] = was_dispatched; + Ok((idx, dispatch_state, Ok(Err(err)))) => { + dispatch_states[idx] = dispatch_state; errs[idx] = Some(err); } - Ok((idx, was_dispatched, Err(_))) => { - dispatched[idx] = was_dispatched; + Ok((idx, dispatch_state, Err(_))) => { + dispatch_states[idx] = dispatch_state; errs[idx] = Some(DiskError::Unexpected); fanout_panic += 1; } @@ -4350,7 +4369,7 @@ impl SetDisks { &coordinator_disks, file_infos, &errs, - &dispatched, + &dispatch_states, &data_dirs, (&fanout_dst_bucket, &fanout_dst_object), rollback_receipt, @@ -4566,7 +4585,7 @@ impl SetDisks { let publication_scope = scanner_publication_commit_scope.clone(); async move { - let mut dispatched = false; + let mut dispatch_state = RenameDispatchState::NotDispatched; let result = std::panic::AssertUnwindSafe(async { // Test-only introspection guard: counts this operation as // in-flight for the whole body. Compiles to `()` in production. @@ -4608,9 +4627,9 @@ impl SetDisks { } let disk_wait_started = rustfs_io_metrics::put_stage_timer(); - dispatched = true; - let result = disk - .rename_data_borrowed_with_fence( + dispatch_state = RenameDispatchState::MayHavePublished; + let observed = disk + .rename_data_borrowed_with_fence_observed( &src_bucket, &src_object, file_info, @@ -4619,6 +4638,8 @@ impl SetDisks { scanner_publication_lease_token, ) .await; + let rejected_before_publication = observed.rejected_before_publication(); + let result = observed.result; #[cfg(test)] if result.is_ok() { rollback_fault_injection::after_rename(&dst_object, i)?; @@ -4644,11 +4665,14 @@ impl SetDisks { }; rustfs_io_metrics::record_put_rename_disk_wait_completion(position, duration_ms); } + if rejected_before_publication { + dispatch_state = RenameDispatchState::RejectedBeforePublication; + } result }) .catch_unwind() .await; - (dispatched, result) + (dispatch_state, result) } }); let results = join_all(futures).await; @@ -4695,9 +4719,9 @@ impl SetDisks { ); } - let mut dispatched = Vec::with_capacity(results.len()); - for (idx, (was_dispatched, result)) in results.iter().enumerate() { - dispatched.push(*was_dispatched); + let mut dispatch_states = Vec::with_capacity(results.len()); + for (idx, (dispatch_state, result)) in results.iter().enumerate() { + dispatch_states.push(*dispatch_state); match result { Ok(Ok(res)) => { data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir); @@ -4763,7 +4787,7 @@ impl SetDisks { disks, file_infos, &errs, - &dispatched, + &dispatch_states, &data_dirs, (&dst_bucket, &dst_object), rollback_receipt, @@ -6727,6 +6751,7 @@ pub(in crate::set_disk) mod rollback_fault_injection { Io, Panic, IoAfterRename, + VolumeNotFoundAfterRename, PanicAfterRename, CoordinatorPanic, } @@ -6775,6 +6800,7 @@ pub(in crate::set_disk) mod rollback_fault_injection { .copied(); match fault { Some((target, Fault::IoAfterRename)) if target == disk_index => Err(DiskError::FaultyDisk), + Some((target, Fault::VolumeNotFoundAfterRename)) if target == disk_index => Err(DiskError::VolumeNotFound), Some((target, Fault::PanicAfterRename)) if target == disk_index => panic!("injected panic after rename mutation"), _ => Ok(()), } @@ -10599,6 +10625,7 @@ mod tests { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { for fault in [ rollback_fault_injection::Fault::IoAfterRename, + rollback_fault_injection::Fault::VolumeNotFoundAfterRename, rollback_fault_injection::Fault::PanicAfterRename, ] { let bucket = "rename-tail-unknown"; @@ -10666,6 +10693,7 @@ mod tests { for fault in [ rollback_fault_injection::Fault::Io, rollback_fault_injection::Fault::IoAfterRename, + rollback_fault_injection::Fault::VolumeNotFoundAfterRename, rollback_fault_injection::Fault::PanicAfterRename, rollback_fault_injection::Fault::CoordinatorPanic, ] { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ce9f95d46..c76daf707 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -17505,27 +17505,69 @@ mod put_object_tmp_cleanup_tests { } #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] async fn put_object_failure_cleans_tmp_workspace_inline() { - let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await; + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] { + let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "tmp-clean-missing-bucket"; + let object = "orphan-object"; + let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace); + let writer = Arc::clone(&set_disks); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("missing-bucket PUT must stage before rename"); + let staged = non_trash_tmp_entries(&temp_dirs).await; + assert_eq!(staged.len(), 4, "every disk must have a staged workspace before rejection"); + for workspace in staged { + let mut entries = tokio::fs::read_dir(&workspace) + .await + .expect("staged workspace should be readable"); + let mut shards = 0; + while let Some(entry) = entries.next_entry().await.expect("staged data directory should be readable") { + if entry.file_type().await.expect("staged entry type").is_dir() { + let part = tokio::fs::metadata(entry.path().join("part.1")) + .await + .expect("staging must contain an actual erasure shard"); + assert!(part.len() > 0, "the shard must be written before the missing-bucket failure"); + shards += 1; + } + } + assert_eq!(shards, 1); + } + assert!(temp_dirs.iter().all(|dir| !dir.path().join(bucket).exists())); + barrier.release(); + let err = tokio::time::timeout(Duration::from_secs(30), put) + .await + .expect("missing-bucket PUT must finish") + .expect("PUT task should join") + .expect_err("put_object into a missing bucket volume must fail"); + assert!(matches!(err, StorageError::VolumeNotFound), "original disk error expected: {err}"); - // 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); + // No polling: known pre-publication rejection must clean staging + // inline, before PUT returns (backlog#864 / backlog#898). + 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}" + ); + } + }) + .await; } #[tokio::test] @@ -18373,87 +18415,92 @@ mod put_object_tmp_cleanup_tests { temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] { - let (dirs, disks, set) = hermetic_set_disks(4).await; - let bucket = "put-incomplete-undo"; - let object = "incomplete-undo-object"; - make_completion_test_bucket(&disks, bucket).await; - let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]); - set.put_object( - bucket, - object, - &mut old_reader, - &ObjectOptions { - write_completion: WriteCompletion::TailDrained, - ..Default::default() - }, - ) - .await - .expect("old generation should be completely committed"); - wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await; - let old = disks[0] - .read_version("", bucket, object, "", &ReadOptions::default()) + for fault in [ + rollback_fault_injection::Fault::Io, + rollback_fault_injection::Fault::VolumeNotFoundAfterRename, + ] { + let (dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = "put-incomplete-undo"; + let object = "incomplete-undo-object"; + make_completion_test_bucket(&disks, bucket).await; + let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]); + set.put_object( + bucket, + object, + &mut old_reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) .await - .expect("old metadata must be readable"); - let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory"); - let tasks = rename_fanout_barrier::observe_tasks(object); - let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); - let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]); - let _undo_fault = rollback_fault_injection::arm(object, 0, rollback_fault_injection::Fault::Io); - let writer = Arc::clone(&set); - let put = tokio::spawn(async move { - let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); - writer - .put_object( - bucket, - object, - &mut reader, - &ObjectOptions { - write_completion, - ..Default::default() - }, - ) + .expect("old generation should be completely committed"); + wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await; + let old = disks[0] + .read_version("", bucket, object, "", &ReadOptions::default()) .await - }); - tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) - .await - .expect("overwrite must enter the actual rename fan-out before failure injection"); - barrier.release(); - let err = tokio::time::timeout(Duration::from_secs(30), put) - .await - .expect("incomplete undo must return without hanging") - .expect("PUT task should join") - .expect_err("two renamed disks cannot satisfy write quorum three"); - assert!( - matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)), - "original quorum error expected: {err}" - ); - assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return"); - let leftovers = non_trash_tmp_entries(&dirs).await; - assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery"); - let backups = dirs - .iter() - .filter(|dir| { - dir.path() - .join(bucket) - .join(object) - .join(old_data_dir.to_string()) - .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP) - .exists() - }) - .count(); - assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup"); - // The remaining three disks still serve the old generation; - // the failed minority must never become an acknowledged write. - let mut read = set - .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) - .await - .expect("old generation must remain readable after incomplete rollback"); - let mut body = Vec::new(); - read.stream - .read_to_end(&mut body) - .await - .expect("old generation should stream"); - assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]); + .expect("old metadata must be readable"); + let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory"); + let tasks = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]); + let _undo_fault = rollback_fault_injection::arm(object, 0, fault); + let writer = Arc::clone(&set); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("overwrite must enter the actual rename fan-out before failure injection"); + barrier.release(); + let err = tokio::time::timeout(Duration::from_secs(30), put) + .await + .expect("incomplete undo must return without hanging") + .expect("PUT task should join") + .expect_err("two renamed disks cannot satisfy write quorum three"); + assert!( + matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)), + "original quorum error expected: {err}" + ); + assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return"); + let leftovers = non_trash_tmp_entries(&dirs).await; + assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery"); + let backups = dirs + .iter() + .filter(|dir| { + dir.path() + .join(bucket) + .join(object) + .join(old_data_dir.to_string()) + .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP) + .exists() + }) + .count(); + assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup"); + // The remaining three disks still serve the old generation; + // the failed minority must never become an acknowledged write. + let mut read = set + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("old generation must remain readable after incomplete rollback"); + let mut body = Vec::new(); + read.stream + .read_to_end(&mut body) + .await + .expect("old generation should stream"); + assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]); + } } }) .await;