mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
fix(ecstore): defer multipart part.N.meta cleanup until after commit (#4548)
complete_multipart_upload deleted every part.N.meta via cleanup_multipart_path *before* the authoritative rename_data commit. If rename_data then failed write quorum, the upload directory was left with data files but no part metadata, so a retried CompleteMultipartUpload could never read the parts again and the upload became permanently uncompletable (client must abort and re-upload all parts). Move cleanup_multipart_path to run only after rename_data returns Ok, matching the existing "clean up only after commit" pattern already used for the old data-dir GC and the upload-dir delete_all. On a failed commit the staging part.N.meta now survive so the retry can complete. Add a hermetic regression test that forces rename_data to fail on every disk (destination bucket dir made read-only) and asserts the staging part.N.meta are preserved for retry. Fixes #946. Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
@@ -1419,7 +1419,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
}
|
||||
|
||||
let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
|
||||
self.cleanup_multipart_path(&parts).await;
|
||||
|
||||
let (online_disks, versions, op_old_dir, cleanup_disks) = Self::rename_data(
|
||||
&shuffle_disks,
|
||||
@@ -1432,6 +1431,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
)
|
||||
.await?;
|
||||
|
||||
// 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
|
||||
// fails write quorum and returns via `?`, the upload directory must keep
|
||||
// its part.N.meta so a retried CompleteMultipartUpload can still read the
|
||||
// parts; deleting them before the commit would strand the upload
|
||||
// permanently. This mirrors the "clean up only after commit" pattern
|
||||
// already used for the old data-dir GC and the upload-dir delete_all below.
|
||||
self.cleanup_multipart_path(&parts).await;
|
||||
|
||||
if let Some(old_dir) = op_old_dir {
|
||||
let committed_dir = fi.data_dir.unwrap_or_default().to_string();
|
||||
// backlog#898: best-effort reclaim of the dereferenced old data dir.
|
||||
@@ -1842,6 +1851,135 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Recursively collect every file named `file_name` under the multipart
|
||||
/// staging bucket on a single disk. Used to observe whether a failed commit
|
||||
/// left the per-part metadata intact for a retry.
|
||||
async fn multipart_meta_files_on_disk(temp_dir: &TempDir, file_name: &str) -> Vec<String> {
|
||||
let mut found = Vec::new();
|
||||
let mut stack = vec![temp_dir.path().join(RUSTFS_META_MULTIPART_BUCKET)];
|
||||
while let Some(dir) = stack.pop() {
|
||||
let mut read_dir = match tokio::fs::read_dir(&dir).await {
|
||||
Ok(read_dir) => read_dir,
|
||||
Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
|
||||
Err(err) => panic!("multipart dir {dir:?} should be listable: {err}"),
|
||||
};
|
||||
while let Some(entry) = read_dir.next_entry().await.expect("multipart dir entry should be readable") {
|
||||
let file_type = entry.file_type().await.expect("dir entry file type should be readable");
|
||||
if file_type.is_dir() {
|
||||
stack.push(entry.path());
|
||||
} else if entry.file_name().to_string_lossy() == file_name {
|
||||
found.push(entry.path().display().to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
found
|
||||
}
|
||||
|
||||
/// Total number of `part.1.meta` files across every disk's multipart staging area.
|
||||
async fn total_part1_meta(temp_dirs: &[TempDir]) -> usize {
|
||||
let mut n = 0;
|
||||
for dir in temp_dirs {
|
||||
n += multipart_meta_files_on_disk(dir, "part.1.meta").await.len();
|
||||
}
|
||||
n
|
||||
}
|
||||
|
||||
/// backlog#946: `complete_multipart_upload` must clean up the stale
|
||||
/// `part.N.meta` files only *after* the authoritative `rename_data` commit
|
||||
/// succeeds. If the commit fails write quorum, the upload directory has to
|
||||
/// keep its `part.N.meta` so a retried CompleteMultipartUpload can still read
|
||||
/// the parts; deleting them beforehand permanently strands the upload.
|
||||
///
|
||||
/// The commit is forced to fail *at rename_data* (not earlier): the multipart
|
||||
/// staging dir stays readable on every disk so `check_upload_id_exists` and
|
||||
/// `read_parts` pass write/read quorum, while the destination bucket
|
||||
/// directory is made unwritable on *every* disk so `rename_data` fails the
|
||||
/// commit on all disks (never moving the staging data) and returns below
|
||||
/// write quorum. That reproduces the issue's real failure mode, where the
|
||||
/// staging `part.N.meta` would still be present if not deleted prematurely.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn complete_multipart_keeps_part_meta_when_commit_fails_for_retry() {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
|
||||
let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
|
||||
let bucket = "multipart-commit-retry-bucket";
|
||||
let object = "object";
|
||||
for disk in &disk_stores {
|
||||
disk.make_volume(bucket).await.expect("bucket volume should be created");
|
||||
}
|
||||
|
||||
let upload = set_disks
|
||||
.new_multipart_upload(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("multipart upload should be created");
|
||||
|
||||
// Upload one full, valid part while every disk is online. A single part
|
||||
// is the last part, so the minimum-part-size gate does not apply and a
|
||||
// small payload keeps the test fast.
|
||||
let part_size = 4096usize;
|
||||
let mut reader = PutObjReader::new(
|
||||
HashReader::from_stream(Cursor::new(vec![9u8; part_size]), part_size as i64, part_size as i64, None, None, false)
|
||||
.expect("hash reader should be constructed"),
|
||||
);
|
||||
let part_info = set_disks
|
||||
.put_object_part(bucket, object, &upload.upload_id, 1, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("uploading the part should succeed");
|
||||
|
||||
// Every disk must hold part.1.meta in the staging area now.
|
||||
let before = total_part1_meta(&temp_dirs).await;
|
||||
assert_eq!(before, temp_dirs.len(), "the uploaded part must write part.1.meta to every disk");
|
||||
|
||||
// Lock the destination bucket directory on *all* disks so the commit's
|
||||
// rename into `bucket/object` fails everywhere: rename_data cannot reach
|
||||
// write quorum and no disk moves the readable staging metadata.
|
||||
let locked_buckets: Vec<std::path::PathBuf> = temp_dirs.iter().map(|d| d.path().join(bucket)).collect();
|
||||
for path in &locked_buckets {
|
||||
let mut perms = std::fs::metadata(path).expect("bucket dir metadata").permissions();
|
||||
perms.set_mode(0o555);
|
||||
std::fs::set_permissions(path, perms).expect("bucket dir should be made read-only");
|
||||
}
|
||||
|
||||
let completed = set_disks
|
||||
.clone()
|
||||
.complete_multipart_upload(
|
||||
bucket,
|
||||
object,
|
||||
&upload.upload_id,
|
||||
vec![CompletePart {
|
||||
part_num: part_info.part_num,
|
||||
etag: part_info.etag.clone(),
|
||||
..Default::default()
|
||||
}],
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await;
|
||||
|
||||
// Restore write permission so the temp dirs can be cleaned up and the
|
||||
// staging metadata re-read below.
|
||||
for path in &locked_buckets {
|
||||
let mut perms = std::fs::metadata(path).expect("bucket dir metadata").permissions();
|
||||
perms.set_mode(0o755);
|
||||
std::fs::set_permissions(path, perms).expect("bucket dir should be made writable again");
|
||||
}
|
||||
|
||||
assert!(
|
||||
completed.is_err(),
|
||||
"commit must fail once the destination is unwritable on every disk, got {completed:?}"
|
||||
);
|
||||
|
||||
// Regression assertion: the failed commit must NOT have deleted the
|
||||
// staging part.1.meta. Old code cleaned up before rename_data and would
|
||||
// leave zero here, permanently breaking a retried CompleteMultipartUpload.
|
||||
let after = total_part1_meta(&temp_dirs).await;
|
||||
assert_eq!(
|
||||
after,
|
||||
temp_dirs.len(),
|
||||
"part.1.meta must survive a failed commit so CompleteMultipartUpload stays retryable"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reduce_quorum_part_numbers_only_keeps_parts_present_on_quorum_of_drives() {
|
||||
let object_parts = vec![
|
||||
|
||||
Reference in New Issue
Block a user