From 3e5d4ebb09ac7437f4a9a462d67a3511d103ecb8 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 5 Sep 2026 20:20:50 +0800 Subject: [PATCH] fix(ecstore): release multipart disk snapshot before nested reads (#7184) * fix(ecstore): release multipart disk snapshot before nested reads * fix(ecstore): remove duplicate local rename implementation Keep the canonical commit module after concurrent storage changes merged. The control-write and rollback changes are already present there. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(app): simplify absent SSE configuration matching * fix(tests): satisfy new clippy lints --------- Co-authored-by: houseme Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/ecstore/src/set_disk/ops/multipart.rs | 88 +++++++++++++++++++- 1 file changed, 84 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index d7d3a39ad..aaf6204b5 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -2452,10 +2452,9 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks { let write_quorum = fi.write_quorum(self.default_write_quorum()); let read_quorum = fi.read_quorum(self.default_read_quorum()); - let disks = self.disks.read().await; - - let disks = disks.clone(); - // let disks = Self::shuffle_disks(&disks, &fi.erasure.distribution); + // Release the registry guard before recovery and cleanup read it again: + // a queued topology writer would otherwise deadlock those nested reads. + let disks = self.get_disks_internal().await; let part_path = format!("{}/{}/", upload_id_path, fi.data_dir.unwrap_or(Uuid::nil())); self.recover_part_transactions(&part_path, read_quorum, write_quorum) @@ -6743,6 +6742,87 @@ mod tests { .await; } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn complete_multipart_releases_disk_snapshot_before_cleanup() { + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "multipart-topology-lock-bucket"; + let object = "object"; + let body = vec![0x65; 4096]; + make_bucket_on_all(&disk_stores, bucket).await; + let (upload_id, parts) = + stage_upload_with_create_opts(&set_disks, bucket, object, &body, &ObjectOptions::default()).await; + let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id); + for dir in &temp_dirs { + assert!( + dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_id_path).exists(), + "the test must create real upload staging on every disk" + ); + } + let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterObjectPublication); + let complete_store = set_disks.clone(); + let complete_upload_id = upload_id.clone(); + let complete = tokio::spawn(async move { + complete_store + .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default()) + .await + }); + barrier.wait_until_paused().await; + + // Hold a separate read gate so the real writer queues even when completion + // correctly releases its snapshot guard. Polling Pending proves admission + // to Tokio's write-preferring queue before the cleanup attempts another read. + let read_gate = set_disks.disks.read().await; + let writer = set_disks.disks.write(); + tokio::pin!(writer); + assert!(matches!( + futures::poll!(tokio::task::unconstrained(writer.as_mut())), + std::task::Poll::Pending + )); + assert!( + set_disks.disks.try_read().is_err(), + "the pending writer must already block new readers before the cleanup resumes" + ); + drop(read_gate); + barrier.release(); + + let writer_guard = tokio::time::timeout(Duration::from_secs(5), writer) + .await + .expect("a queued topology writer must not deadlock with multipart cleanup's disk snapshot"); + // A reconnect can publish the same handles; this test isolates admission + // order without changing the disks that contain the committed object. + drop(writer_guard); + tokio::time::timeout(Duration::from_secs(10), complete) + .await + .expect("multipart cleanup must finish after the topology writer releases") + .expect("completion task should not panic") + .expect("completion should preserve the successful object commit"); + + let mut reader = tokio::time::timeout( + Duration::from_secs(10), + set_disks.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()), + ) + .await + .expect("GET should finish after completion") + .expect("the completed object should remain readable"); + let mut observed_body = Vec::new(); + tokio::time::timeout(Duration::from_secs(10), reader.stream.read_to_end(&mut observed_body)) + .await + .expect("the completed object body should finish streaming") + .expect("the completed object body should be readable"); + assert_eq!(observed_body, body); + assert!(matches!( + set_disks.check_upload_id_exists(bucket, object, &upload_id, false).await, + Err(StorageError::InvalidUploadID(..)) + )); + for dir in &temp_dirs { + assert!( + !dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_id_path).exists(), + "successful completion must remove its upload staging from every disk" + ); + } + } + #[tokio::test(flavor = "multi_thread")] #[serial] async fn complete_releases_object_lock_before_cleanup_and_keeps_upload_lock() {