From f78f146c358b42e2eb1acdb6f8b3595e08b9e892 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 15 Jul 2026 16:10:14 +0800 Subject: [PATCH] test(ecstore): extend crash-point injection to multipart complete and xl.meta update paths (#4853) --- .config/nextest.toml | 16 ++ crates/ecstore/src/crash_inject.rs | 146 ++++++++++++ crates/ecstore/src/disk/local.rs | 227 +++++++++++------- crates/ecstore/src/lib.rs | 1 + crates/ecstore/src/set_disk/ops/multipart.rs | 238 +++++++++++++++++++ 5 files changed, 542 insertions(+), 86 deletions(-) create mode 100644 crates/ecstore/src/crash_inject.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 965523274..bebd726db 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -45,6 +45,15 @@ e2e-reliability = { max-threads = 1 } filter = 'package(rustfs-ecstore) & (test(concurrent_resend_same_part_commits_one_generation) | test(/^store::bucket::tests::bucket_delete_(mark_delete_marks|purge_removes|default_s3_delete)/))' test-group = 'ecstore-serial-flaky' +# Serialize the multipart crash-consistency scenarios (dist-2, backlog#1150): +# each spawns a 4-disk hermetic erasure set and drives full staged-upload + +# commit + GET cycles — the same cross-disk-commit IO shape that made +# concurrent_resend load-sensitive. Preventive serialization only, no retries. +# The matching ci-profile override is after [profile.ci]. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)' +test-group = 'ecstore-serial-flaky' + # Serialize the 4-disk reliability / degraded-read e2e tests (see the # e2e-reliability test-group note above). The matching ci-profile override is at # the end of the file, after [profile.ci] is declared. @@ -109,6 +118,13 @@ retries = 2 filter = 'package(e2e_test) & test(/^(reliability_disk_fault|degraded_read_eof_regression)_test::/)' test-group = 'e2e-reliability' +# Serialize the multipart crash-consistency scenarios under the ci profile too +# (see the matching default-profile override near the top). Not a quarantine: +# no retries, just serialized 4-disk cross-disk-commit IO. +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(/^set_disk::ops::multipart::tests::crash_consistency::/)' +test-group = 'ecstore-serial-flaky' + # --------------------------------------------------------------------------- # e2e-smoke profile — PR smoke subset of the e2e_test crate (backlog#1149 ci-4) # --------------------------------------------------------------------------- diff --git a/crates/ecstore/src/crash_inject.rs b/crates/ecstore/src/crash_inject.rs new file mode 100644 index 000000000..30abe0d70 --- /dev/null +++ b/crates/ecstore/src/crash_inject.rs @@ -0,0 +1,146 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Deterministic crash-point injection for erasure-coded commit sequences. +//! +//! Each [`CrashPoint`] names one instant inside a multi-step commit at which a +//! `#[cfg(test)]` scenario can simulate a hard power loss: the commit stops dead +//! at the armed step with **no** cleanup, leaving the on-disk state exactly as +//! the preceding steps left it. The test then reopens the disk (or rebuilds the +//! erasure set) and asserts the raw state is coherent — the object reads back as +//! either the whole old version or the whole new version, never a torn mix — and +//! that any staged leftovers are reclaimable and the operation is safely +//! retryable. This is the crash-consistency counterpart to the graceful +//! `should_fail_*` failpoints, which exercise in-process rollback rather than an +//! abrupt loss. +//! +//! Unlike the graceful failpoints, a crash point runs no rollback: it models the +//! process disappearing, so the assertion is purely over the bytes left on disk. +//! +//! Test-only: the arming registry and the real [`should_crash_at`] are +//! `#[cfg(test)]`. In every other build `should_crash_at` is a const-`false` +//! `#[inline(always)]` no-op, so the injection points on the real commit paths +//! (`rename_data`, `write_all_meta`, `complete_multipart_upload`) compile away to +//! nothing and add no branch to production writes. The [`CrashPoint`] variants +//! are still constructed at those call sites in every build, so no variant is +//! dead code. +//! +//! Coverage plan: rustfs/backlog#935 / rustfs/backlog#896 (rename_data), +//! rustfs/backlog#864 (fault-interrupted overwrite/delete must not mutate a +//! committed object), rustfs/backlog#878 (old-or-new-never-mixed). + +/// A named instant inside an erasure-coded commit sequence at which a test can +/// simulate a hard power loss. +/// +/// All points share one keyed registry (each arm is matched on `(point, key)`), +/// so a scenario arms exactly the window under test and every other commit path +/// runs untouched. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum CrashPoint { + /// `rename_data`: after the data dir is renamed into its destination but + /// before the old-metadata rollback backup is written. Pre-commit — xl.meta + /// is untouched, so a crash here must leave the object readable as the old + /// version (or absent when there was none). + RenameAfterDataRename, + /// `rename_data`: after the rollback backup is durable, immediately before + /// the xl.meta commit rename that makes the new version visible. Still + /// pre-commit, so a crash here must also leave the old version readable. + RenameAfterBackupBeforeMetaCommit, + /// `rename_data`: after the xl.meta commit rename has made the new version + /// visible, before the durability fsync and with **no** rollback. The commit + /// rename already landed, so a crash here must leave the object readable as + /// the new version (rustfs/backlog#878, new-version side). + RenameAfterMetaCommit, + /// `write_all_meta` (the atomic temp+rename behind `update_metadata` and + /// `write_metadata`): after the replacement xl.meta is staged in the tmp + /// bucket but before the rename that publishes it. Pre-commit — the + /// destination xl.meta is untouched, so a crash here must leave the object's + /// metadata byte-for-byte the old version (rustfs/backlog#864); the staged + /// tmp file is a harmless orphan swept by tmp-bucket GC. + MetaWriteAfterTmpBeforeRename, + /// `complete_multipart_upload`: after the upload is fully staged and locked + /// but before the authoritative `rename_data` commit. Pre-commit — no disk + /// has moved staged data, so a crash here must leave any prior committed + /// version byte-for-byte intact (rustfs/backlog#864) and the upload fully + /// retryable. + MultipartBeforeCommitRename, + /// `complete_multipart_upload`: after the `rename_data` commit succeeds but + /// before the stale `part.N.meta` cleanup. Post-commit — the new version is + /// durably committed and visible, so a crash here must leave the object + /// readable as the new version; the un-reclaimed staging parts are swept by + /// a retried completion or upload GC (rustfs/backlog#946). + MultipartAfterCommitBeforePartsCleanup, +} + +#[cfg(test)] +pub(crate) use armed::{arm, disarm, should_crash_at}; + +/// Production no-op: the compiler inlines this to `false`, so armed crash points +/// exist only in `#[cfg(test)]` builds and never add a branch to real commits. +#[cfg(not(test))] +#[inline(always)] +pub(crate) fn should_crash_at(_point: CrashPoint, _key: &str) -> bool { + false +} + +#[cfg(test)] +mod armed { + use super::CrashPoint; + use std::sync::{Mutex, MutexGuard}; + + // A keyed multiset of pending arms, not a single slot: the rename_data / + // xl.meta scenarios serialize via `durability_mode_override` while the + // multipart scenarios use `#[serial]`, so the two groups can overlap under + // `cargo test`'s shared-process threads. Keying every arm on a unique object + // path lets concurrent scenarios coexist without clobbering each other; a + // match on `(point, key)` is unambiguous. `#[serial]` still gives the + // multipart tests their own serialization for unrelated global state. + static ARMED: Mutex> = Mutex::new(Vec::new()); + + fn guard() -> MutexGuard<'static, Vec<(CrashPoint, String)>> { + // A poisoned lock only means a prior test panicked mid-scenario; recover + // the registry rather than cascade the panic into unrelated tests. + ARMED.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + /// Arm a one-shot crash: the next operation that reaches `point` while + /// committing `key` stops there. Consumed on the first match so it never + /// leaks into an unrelated commit. Arming the same `(point, key)` twice is + /// idempotent. + pub(crate) fn arm(point: CrashPoint, key: &str) { + let mut armed = guard(); + if !armed.iter().any(|(p, k)| *p == point && k == key) { + armed.push((point, key.to_string())); + } + } + + /// Clear the pending arm for exactly this `(point, key)`. A scenario retries + /// the same key after the crash, so teardown disarms in case the injection + /// was somehow never reached and would otherwise fire inside the retry. + pub(crate) fn disarm(point: CrashPoint, key: &str) { + guard().retain(|(p, k)| !(*p == point && k == key)); + } + + /// Returns `true` (consuming the arm) iff a crash is armed for exactly this + /// `point` and `key`. + pub(crate) fn should_crash_at(point: CrashPoint, key: &str) -> bool { + let mut armed = guard(); + if let Some(idx) = armed.iter().position(|(p, k)| *p == point && k == key) { + armed.swap_remove(idx); + true + } else { + false + } + } +} diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 255bd6a93..dd5dc8c67 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::config::storageclass::DEFAULT_INLINE_BLOCK; +use crate::crash_inject::{self, CrashPoint}; use crate::data_usage::local_snapshot::ensure_data_usage_layout; use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout}; use crate::disk::{ @@ -1478,79 +1479,6 @@ fn should_fail_before_old_metadata_backup(_dst_path: &str) -> bool { false } -/// Commit-sequence points where the rename_data crash-consistency harness -/// (rustfs/backlog#935, test plan in rustfs/backlog#896) can simulate an -/// abrupt power loss. -/// -/// Unlike [`should_fail_before_old_metadata_backup`], which exercises the -/// graceful in-process rollback (delete the staged data dir, return an -/// error), a crash point models a hard power loss: the commit sequence stops -/// dead at the armed step with **no** cleanup, leaving the on-disk state -/// exactly as the preceding steps left it. The harness then reopens the disk -/// and asserts the raw state is coherent — the object reads back as either the -/// old version or the new version, never a mixed or corrupt one — without any -/// rollback code having run. -/// -/// The variants are constructed at the real commit-path call sites in every -/// build, but the arming static and [`should_crash_rename_data_at`] are -/// `#[cfg(test)]`; in production the guard is a const-`false` no-op, so the -/// injection points compile away to nothing. -// The `After` naming is deliberate: every crash point names the -// commit-sequence step it fires immediately after, so the shared prefix is the -// point, not noise. -#[allow(clippy::enum_variant_names)] -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub(crate) enum RenameDataCrashPoint { - /// After the data dir has been renamed into its destination but before the - /// old-metadata rollback backup is written. xl.meta has not been committed - /// yet, so a crash here must leave the object readable as the old version. - AfterDataRename, - /// After the rollback backup is persisted, immediately before the xl.meta - /// commit rename that makes the new version visible. Still pre-commit, so a - /// crash here must also leave the object readable as the old version. - AfterBackupBeforeMetaCommit, - /// After the xl.meta commit rename has made the new version visible, in the - /// same window the graceful [`should_fail_after_metadata_commit`] failpoint - /// covers — but as a hard power loss with **no** rollback. The commit rename - /// already landed on disk, so a crash here must leave the object readable as - /// the *new* version. This is the post-commit counterpart to the two - /// pre-commit points above, closing the "old or new, never mixed" invariant - /// (rustfs/backlog#878) from the new-version side. - AfterMetaCommit, -} - -#[cfg(test)] -static RENAME_DATA_CRASH_POINT: std::sync::Mutex> = std::sync::Mutex::new(None); - -/// Arm a one-shot crash injection: the next `rename_data` committing into -/// `dst_path` stops at `point`. Consumed on the first match so it never leaks -/// into an unrelated commit. -#[cfg(test)] -fn arm_rename_data_crash(point: RenameDataCrashPoint, dst_path: &str) { - *RENAME_DATA_CRASH_POINT - .lock() - .expect("test crash point lock should not be poisoned") = Some((point, dst_path.to_string())); -} - -#[cfg(test)] -fn should_crash_rename_data_at(point: RenameDataCrashPoint, dst_path: &str) -> bool { - let mut armed = RENAME_DATA_CRASH_POINT - .lock() - .expect("test crash point lock should not be poisoned"); - if armed.as_ref().is_some_and(|(p, path)| *p == point && path == dst_path) { - armed.take(); - true - } else { - false - } -} - -#[cfg(not(test))] -#[inline(always)] -fn should_crash_rename_data_at(_point: RenameDataCrashPoint, _dst_path: &str) -> bool { - false -} - #[cfg(not(test))] fn should_fail_after_metadata_commit(_dst_path: &str) -> bool { false @@ -4870,6 +4798,16 @@ impl LocalDisk { self.write_all_internal(&tmp_file_path, InternalBuf::Ref(buf), tmp_sync, &tmp_volume_dir) .await?; + // Crash-consistency injection: hard power loss after the replacement + // xl.meta is staged in the tmp bucket but before the atomic rename that + // publishes it. The destination xl.meta is untouched, so a crash here + // must leave the object's metadata byte-for-byte the old version + // (rustfs/backlog#864); the staged tmp file is a harmless orphan swept by + // tmp-bucket GC. Compiles to a no-op outside `#[cfg(test)]`. + if crash_inject::should_crash_at(CrashPoint::MetaWriteAfterTmpBeforeRename, path) { + return Err(DiskError::Unexpected); + } + rename_all(tmp_file_path, &file_path, volume_dir).await?; if sync @@ -6853,7 +6791,7 @@ impl DiskAPI for LocalDisk { // is in place but before xl.meta commits. No cleanup — the harness // reopens the disk and asserts the object still reads as the old // version (the staged data dir is a harmless orphan for GC). - if should_crash_rename_data_at(RenameDataCrashPoint::AfterDataRename, dst_path) { + if crash_inject::should_crash_at(CrashPoint::RenameAfterDataRename, dst_path) { return Err(DiskError::Unexpected); } @@ -6911,7 +6849,7 @@ impl DiskAPI for LocalDisk { // backup is durable but before the xl.meta commit rename. No // cleanup — the harness asserts the object still reads as the old // version, since the destination xl.meta is untouched here. - if should_crash_rename_data_at(RenameDataCrashPoint::AfterBackupBeforeMetaCommit, dst_path) { + if crash_inject::should_crash_at(CrashPoint::RenameAfterBackupBeforeMetaCommit, dst_path) { return Err(DiskError::Unexpected); } @@ -6944,7 +6882,7 @@ impl DiskAPI for LocalDisk { // graceful failpoint above, no rollback runs — the commit rename is // already on disk, so the harness asserts the object reads back as // the new version. - if should_crash_rename_data_at(RenameDataCrashPoint::AfterMetaCommit, dst_path) { + if crash_inject::should_crash_at(CrashPoint::RenameAfterMetaCommit, dst_path) { return Err(DiskError::Unexpected); } @@ -8184,13 +8122,14 @@ mod test { /// old-or-new invariant, only with a wider (documented) power-loss window. mod crash_consistency { use super::*; + use crate::crash_inject::{self, CrashPoint}; use tempfile::tempdir; const VERSION_ID: &str = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"; const OLD_DATA_DIR: &str = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"; const NEW_DATA_DIR: &str = "cccccccc-cccc-cccc-cccc-cccccccccccc"; - async fn run_scenario(mode: DurabilityMode, crash: Option, with_old_version: bool) { + async fn run_scenario(mode: DurabilityMode, crash: Option, with_old_version: bool) { // Serializes with every other durability-sensitive test and pins the // resolved tier for the whole scenario (held until dropped). let _mode = durability_mode_override::set(mode); @@ -8241,7 +8180,7 @@ mod test { .expect("new tmp data should be written"); if let Some(point) = crash { - arm_rename_data_crash(point, object); + crash_inject::arm(point, object); } let new_fi = test_file_info(object, version_id, Some(new_data_dir), None); let result = disk @@ -8256,7 +8195,7 @@ mod test { .await; match crash { - Some(RenameDataCrashPoint::AfterMetaCommit) => { + Some(CrashPoint::RenameAfterMetaCommit) => { // Hard power loss right after the xl.meta commit rename: the // commit landed and no rollback ran, so the object must read // back as the new version — whether or not an old version @@ -8324,9 +8263,9 @@ mod test { } } - const CRASH_POINTS: [RenameDataCrashPoint; 2] = [ - RenameDataCrashPoint::AfterDataRename, - RenameDataCrashPoint::AfterBackupBeforeMetaCommit, + const CRASH_POINTS: [CrashPoint; 2] = [ + CrashPoint::RenameAfterDataRename, + CrashPoint::RenameAfterBackupBeforeMetaCommit, ]; #[tokio::test] @@ -8372,22 +8311,138 @@ mod test { // the same invariant. #[tokio::test] async fn overwrite_post_commit_crash_keeps_new_version_strict() { - run_scenario(DurabilityMode::Strict, Some(RenameDataCrashPoint::AfterMetaCommit), true).await; + run_scenario(DurabilityMode::Strict, Some(CrashPoint::RenameAfterMetaCommit), true).await; } #[tokio::test] async fn overwrite_post_commit_crash_keeps_new_version_relaxed() { - run_scenario(DurabilityMode::Relaxed, Some(RenameDataCrashPoint::AfterMetaCommit), true).await; + run_scenario(DurabilityMode::Relaxed, Some(CrashPoint::RenameAfterMetaCommit), true).await; } #[tokio::test] async fn fresh_post_commit_crash_keeps_new_version_strict() { - run_scenario(DurabilityMode::Strict, Some(RenameDataCrashPoint::AfterMetaCommit), false).await; + run_scenario(DurabilityMode::Strict, Some(CrashPoint::RenameAfterMetaCommit), false).await; } #[tokio::test] async fn fresh_post_commit_crash_keeps_new_version_relaxed() { - run_scenario(DurabilityMode::Relaxed, Some(RenameDataCrashPoint::AfterMetaCommit), false).await; + run_scenario(DurabilityMode::Relaxed, Some(CrashPoint::RenameAfterMetaCommit), false).await; + } + } + + /// Crash-consistency for the in-place xl.meta update path — the atomic + /// temp+rename inside [`LocalDisk::write_all_meta`] shared by `update_metadata` + /// and `write_metadata` (delete markers, tag/metadata rewrites, decommission). + /// + /// rustfs/backlog#864: a fault that interrupts an in-place metadata rewrite + /// must not mutate the committed object. [`CrashPoint::MetaWriteAfterTmpBeforeRename`] + /// models a hard power loss after the replacement xl.meta is staged in the tmp + /// bucket but before the publishing rename. Both durability tiers are held to + /// the same rule: the destination xl.meta survives byte-for-byte, the object + /// still reads as the old version, the staged tmp file is a reclaimable orphan + /// confined to the tmp bucket, and a later un-injected rewrite publishes + /// cleanly (retryable). This path previously had only parser-level unit tests. + mod meta_write_crash_consistency { + use super::*; + use crate::crash_inject::{self, CrashPoint}; + use tempfile::tempdir; + + async fn run(mode: DurabilityMode) { + // Serialize with every other durability-sensitive test and pin the + // resolved tier for the whole scenario. + let _mode = durability_mode_override::set(mode); + + let dir = 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 = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + + let bucket = "bucket"; + let object = "meta-write-crash-object"; + ensure_test_volume(&disk, bucket).await; + ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await; + + // Seed a committed object (version 1) by writing its xl.meta directly: + // read_data=false reads never touch the data dir, so no shard staging + // is needed to exercise the metadata commit window. + let object_dir = dir.path().join(bucket).join(object); + fs::create_dir_all(&object_dir).await.expect("object dir should be created"); + let meta_path = object_dir.join(STORAGE_FORMAT_FILE); + let v1 = Uuid::new_v4(); + let old_meta = test_meta(test_file_info(object, v1, Some(Uuid::new_v4()), None)); + fs::write(&meta_path, &old_meta) + .await + .expect("committed xl.meta should be written"); + + // Arm the crash, then attempt an in-place rewrite that adds version 2. + let meta_key = format!("{object}/{STORAGE_FORMAT_FILE}"); + crash_inject::arm(CrashPoint::MetaWriteAfterTmpBeforeRename, &meta_key); + let v2 = Uuid::new_v4(); + let result = disk + .write_metadata("", bucket, object, test_file_info(object, v2, Some(Uuid::new_v4()), None)) + .await; + assert!( + matches!(result, Err(DiskError::Unexpected)), + "{mode:?}: the armed crash point must be the failure that surfaced, got {result:?}" + ); + + // Reopen the disk to model a process restart after the crash. + drop(disk); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should reopen"); + + // rustfs/backlog#864: the committed xl.meta is byte-for-byte the old + // version — the interrupted rewrite never published. + let after = fs::read(&meta_path).await.expect("xl.meta must survive the crash"); + assert_eq!(&after, &old_meta, "{mode:?}: xl.meta must remain the old version byte-for-byte"); + + // The old version still reads; the un-published new version is absent. + disk.read_version("", bucket, object, &v1.to_string(), &ReadOptions::default()) + .await + .expect("the old version must remain readable after the crash"); + let new_read = disk + .read_version("", bucket, object, &v2.to_string(), &ReadOptions::default()) + .await; + assert!( + matches!(new_read, Err(DiskError::FileVersionNotFound)), + "{mode:?}: the interrupted update must not publish the new version, got {new_read:?}" + ); + + // The crash leaves no staging debris in the object directory; the + // staged replacement is a reclaimable orphan under the tmp bucket. + let mut object_entries = fs::read_dir(&object_dir).await.expect("object dir should list"); + let mut object_files = Vec::new(); + while let Some(entry) = object_entries + .next_entry() + .await + .expect("object dir entry should be readable") + { + object_files.push(entry.file_name().to_string_lossy().to_string()); + } + assert_eq!( + object_files, + vec![STORAGE_FORMAT_FILE.to_string()], + "{mode:?}: the object directory must hold only its committed xl.meta" + ); + + // A retried rewrite (un-injected) publishes cleanly: the path is safely + // retryable after the crash. + crash_inject::disarm(CrashPoint::MetaWriteAfterTmpBeforeRename, &meta_key); + disk.write_metadata("", bucket, object, test_file_info(object, v2, Some(Uuid::new_v4()), None)) + .await + .expect("a retried metadata rewrite must succeed after the crash"); + disk.read_version("", bucket, object, &v2.to_string(), &ReadOptions::default()) + .await + .expect("the retried version must be readable"); + } + + #[tokio::test] + async fn meta_write_crash_before_rename_keeps_old_version_strict() { + run(DurabilityMode::Strict).await; + } + + #[tokio::test] + async fn meta_write_crash_before_rename_keeps_old_version_relaxed() { + run(DurabilityMode::Relaxed).await; } } diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index e6c61e9a1..a30223c03 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -36,6 +36,7 @@ mod cache_value; mod cluster; mod config; mod core; +mod crash_inject; mod data_movement; mod data_usage; mod diagnostics; diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 7053e401d..55be919a4 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -21,6 +21,7 @@ //! unchanged; method bodies are moved verbatim and runtime behavior is the same. use super::super::*; +use crate::crash_inject::{self, CrashPoint}; use std::future::Future; use std::time::Duration; use tokio::task::JoinSet; @@ -1420,6 +1421,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now); + // Crash-consistency injection: hard power loss after the upload is fully + // staged and locked but before the authoritative rename_data commit. No + // disk has moved the staged data, so a crash here must leave any prior + // committed version byte-for-byte intact (rustfs/backlog#864) and the + // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`. + if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, object) { + return Err(StorageError::Unexpected); + } + // The trailing `_` drops the rename_data old-size backfill // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit // `get_object_info` lookup, so the backfill has no consumer here yet. @@ -1434,6 +1444,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { ) .await?; + // Crash-consistency injection: hard power loss after the authoritative + // rename_data commit succeeded but before the stale part.N.meta cleanup. + // The new version is durably committed and visible, so a crash here must + // leave the object readable as the new version; the un-reclaimed staging + // parts are swept by a retried completion or upload GC (rustfs/backlog#946). + // Compiles to a no-op outside `#[cfg(test)]`. + if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object) { + return Err(StorageError::Unexpected); + } + // backlog#946: reclaim the stale per-part metadata (and any superfluous // part.N data files no longer in the completed set) only *after* the // authoritative rename_data commit above has succeeded. If rename_data @@ -2125,4 +2145,222 @@ mod tests { let paged: Vec<&str> = first.iter().chain(second.iter()).map(|u| u.upload_id.as_str()).collect(); assert_eq!(paged, vec!["u0", "u1", "u2", "u3"]); } + + /// Crash-consistency for the two `complete_multipart_upload` commit windows. + /// + /// rustfs/backlog#864: a fault that interrupts a completion must never mutate + /// an already-committed object; the object must read back as the whole old + /// version or the whole new version, never a torn mix, and the upload must + /// stay recoverable. + /// + /// [`CrashPoint::MultipartBeforeCommitRename`] fires after the upload is fully + /// staged and locked but before the authoritative `rename_data` commit: a + /// prior committed version survives byte-for-byte and the upload is retryable. + /// [`CrashPoint::MultipartAfterCommitBeforePartsCleanup`] fires after the + /// commit lands but before the stale `part.N.meta` cleanup: the new version is + /// readable and the un-reclaimed staging is harmless and reclaimable. + mod crash_consistency { + use super::*; + use crate::crash_inject::{self, CrashPoint}; + use crate::storage_api_contracts::object::ObjectIO as _; + use http::HeaderMap; + use tokio::io::AsyncReadExt as _; + + /// 1 MiB keeps every object off the 128 KiB inline fast path, so the + /// commit moves real erasure shards through `rename_data`. + const PART_SIZE: usize = 1 << 20; + + fn payload(fill: u8) -> Vec { + vec![fill; PART_SIZE] + } + + async fn make_bucket_on_all(disks: &[DiskStore], bucket: &str) { + for disk in disks { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + } + + /// Stage a single-part multipart upload without completing it. A single + /// part is the last part, so the 5 MiB minimum-part-size gate does not + /// apply. Returns the upload id and the `CompletePart` list. + async fn stage_upload( + set_disks: &Arc, + bucket: &str, + object: &str, + content: &[u8], + ) -> (String, Vec) { + let upload = set_disks + .new_multipart_upload(bucket, object, &ObjectOptions::default()) + .await + .expect("multipart upload should be created"); + let mut reader = PutObjReader::new( + HashReader::from_stream( + Cursor::new(content.to_vec()), + content.len() as i64, + content.len() as i64, + None, + None, + false, + ) + .expect("hash reader should be constructed"), + ); + let part = set_disks + .put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default()) + .await + .expect("uploading the part should succeed"); + ( + upload.upload_id, + vec![CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }], + ) + } + + async fn complete( + set_disks: &Arc, + bucket: &str, + object: &str, + upload_id: &str, + parts: Vec, + ) -> Result { + set_disks + .clone() + .complete_multipart_upload(bucket, object, upload_id, parts, &ObjectOptions::default()) + .await + } + + /// Read the whole committed object back; returns `(body, etag)`. The full + /// body read also proves the served length matches (never a torn/short body). + async fn read_object(set_disks: &Arc, bucket: &str, object: &str) -> (Vec, Option) { + let mut reader = set_disks + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("object reader should open"); + let etag = reader.object_info.etag.clone(); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("object should stream fully"); + (body, etag) + } + + async fn upload_is_listed(set_disks: &Arc, bucket: &str, object: &str, upload_id: &str) -> bool { + let page = set_disks + .list_multipart_uploads(bucket, object, None, None, None, 1000) + .await + .expect("listing multipart uploads should succeed"); + page.uploads.iter().any(|u| u.upload_id == upload_id) + } + + #[tokio::test] + #[serial] + async fn pre_commit_crash_keeps_committed_old_version() { + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-crash-pre"; + let object = "crash-pre-object"; + make_bucket_on_all(&disk_stores, bucket).await; + + // Commit an OLD version through the multipart path. + let old = payload(0xA1); + let (u_old, parts_old) = stage_upload(&set_disks, bucket, object, &old).await; + complete(&set_disks, bucket, object, &u_old, parts_old) + .await + .expect("the old version should commit"); + let (old_body, old_etag) = read_object(&set_disks, bucket, object).await; + assert_eq!(old_body, old, "the old version must round-trip before the crash"); + + // Stage a replacement upload with NEW content, then crash before commit. + let new = payload(0xB2); + let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; + crash_inject::arm(CrashPoint::MultipartBeforeCommitRename, object); + let crashed = complete(&set_disks, bucket, object, &u_new, parts_new.clone()).await; + assert!( + matches!(crashed, Err(StorageError::Unexpected)), + "the armed pre-commit crash point must be the failure that surfaced, got {crashed:?}" + ); + + // rustfs/backlog#864: the committed old version is untouched, byte-for-byte. + let (after_body, after_etag) = read_object(&set_disks, bucket, object).await; + assert_eq!(after_body, old, "the interrupted completion must not mutate the committed old version"); + assert_eq!(after_etag, old_etag, "the old version's ETag must be unchanged"); + + // The staged upload survived intact (part.N.meta present on quorum), so + // a retried completion commits the new version cleanly (retryable). + assert!( + total_part1_meta(&temp_dirs).await > 0, + "the un-committed upload must keep its staging parts for retry" + ); + crash_inject::disarm(CrashPoint::MultipartBeforeCommitRename, object); + complete(&set_disks, bucket, object, &u_new, parts_new) + .await + .expect("a retried completion must succeed"); + let (retry_body, retry_etag) = read_object(&set_disks, bucket, object).await; + assert_eq!(retry_body, new, "the retried completion must publish the whole new version"); + assert_ne!(retry_etag, old_etag, "the new version must carry a distinct ETag"); + } + + #[tokio::test] + #[serial] + async fn post_commit_crash_publishes_new_version_and_leaves_reclaimable_upload() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-crash-post"; + let object = "crash-post-object"; + make_bucket_on_all(&disk_stores, bucket).await; + + // Stage an upload and crash AFTER the commit rename but BEFORE cleanup. + let new = payload(0xC3); + let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await; + let parts_retry = parts_new.clone(); + crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await; + assert!( + matches!(crashed, Err(StorageError::Unexpected)), + "the armed post-commit crash point must be the failure that surfaced, got {crashed:?}" + ); + crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object); + + // The commit landed: the new version reads back whole and correct. + let (body, _etag) = read_object(&set_disks, bucket, object).await; + assert_eq!(body, new, "a post-commit crash must leave the whole new version readable"); + + // The completion never reached its cleanup, so the upload's staging + // directory is still listable — leftover that must be reclaimable. + assert!( + upload_is_listed(&set_disks, bucket, object, &u_new).await, + "the post-commit crash must leave the upload listable for reclamation" + ); + + // A retried CompleteMultipartUpload is answered deterministically: the + // commit rename already consumed the upload's metadata, so the retry + // resolves to InvalidUploadID (NoSuchUpload to the S3 client, the + // standard answer for a completed-then-retried upload) — never a torn + // state, and the committed object is untouched by the retry. + let retried = complete(&set_disks, bucket, object, &u_new, parts_retry).await; + assert!( + matches!(retried, Err(StorageError::InvalidUploadID(..))), + "a retried complete after the commit landed must resolve to InvalidUploadID, got {retried:?}" + ); + let (body_after_retry, _) = read_object(&set_disks, bucket, object).await; + assert_eq!(body_after_retry, new, "the failed retry must not disturb the committed object"); + + // Reclaim the leftover exactly as the production tail does: delete_all + // on the upload path (abort_multipart_upload cannot — the upload's + // metadata is gone, so it too answers InvalidUploadID). + let upload_dir = SetDisks::get_upload_id_dir(bucket, object, &u_new); + set_disks + .delete_all(RUSTFS_META_MULTIPART_BUCKET, &upload_dir) + .await + .expect("sweeping the leftover upload staging should succeed"); + assert!( + !upload_is_listed(&set_disks, bucket, object, &u_new).await, + "reclaiming the upload must drop it from the listing" + ); + let (body_after, _) = read_object(&set_disks, bucket, object).await; + assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object"); + } + } }