gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 01:09:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): complete rename tails after quorum ack (#6489)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-24 14:30:50 +08:00
committed by GitHub
parent 51272d34dd
commit 86e969f63c
7 changed files with 1543 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/ops/list.rs
+26 -15
View File
@@ -27,13 +27,19 @@ use super::super::*;
impl SetDisks {
#[tracing::instrument(skip(self))]
pub async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
ListOperations::new(self.ctx()).delete_all(bucket, prefix).await
let (result, disks) = ListOperations::new(self.ctx())
.delete_all_observed(bucket, prefix, None)
.await;
self.record_capacity_scope_if_needed(None, &disks);
result
}
pub(crate) async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
ListOperations::new(self.ctx())
.delete_all_with_quorum(bucket, prefix, write_quorum)
.await
let (result, disks) = ListOperations::new(self.ctx())
.delete_all_observed(bucket, prefix, Some(write_quorum))
.await;
self.record_capacity_scope_if_needed(None, &disks);
result
}
}
@@ -53,19 +59,24 @@ impl<'a> ListOperations<'a> {
Self { ctx }
}
pub(crate) async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
self.delete_all_inner(bucket, prefix, None).await
pub(crate) async fn delete_all_observed(
&self,
bucket: &str,
prefix: &str,
write_quorum: Option<usize>,
) -> (Result<()>, Vec<Option<DiskStore>>) {
let disks = self.ctx.disks().read().await.clone();
let result = self.delete_all_inner(bucket, prefix, write_quorum, disks.clone()).await;
(result, disks)
}
async fn delete_all_with_quorum(&self, bucket: &str, prefix: &str, write_quorum: usize) -> Result<()> {
self.delete_all_inner(bucket, prefix, Some(write_quorum)).await
}
async fn delete_all_inner(&self, bucket: &str, prefix: &str, write_quorum: Option<usize>) -> Result<()> {
let disks = self.ctx.disks().read().await;
let disks = disks.clone();
async fn delete_all_inner(
&self,
bucket: &str,
prefix: &str,
write_quorum: Option<usize>,
disks: Vec<Option<DiskStore>>,
) -> Result<()> {
let mut futures = Vec::with_capacity(disks.len());
let mut errors = Vec::with_capacity(disks.len());
crates/ecstore/src/set_disk/ops/locking.rs
+5 -4
View File
@@ -922,10 +922,11 @@ mod tests {
// The List family runs through the borrow handle with unchanged
// behavior: delete_all reports success even when the prefix is absent.
ListOperations::new(set_disks.ctx())
.delete_all("nonexistent-bucket", "nonexistent-prefix")
.await
.expect("delete_all via borrow handle should succeed");
let (result, observed_disks) = ListOperations::new(set_disks.ctx())
.delete_all_observed("nonexistent-bucket", "nonexistent-prefix", None)
.await;
result.expect("delete_all via borrow handle should succeed");
assert_eq!(observed_disks.len(), disk_count);
set_disks
.delete_all("nonexistent-bucket", "nonexistent-prefix")
.await
crates/ecstore/src/set_disk/ops/multipart.rs
+365 -50
View File
@@ -30,6 +30,7 @@ use super::object::{
use crate::bucket::quota::reservation;
use crate::crash_inject::{self, CrashPoint};
use crate::multipart_listing::paginate_multipart_listing;
use crate::set_disk::core::io_primitives::finish_rename_tail_heal;
use futures::{StreamExt, stream};
use std::future::Future;
#[cfg(test)]
@@ -2621,7 +2622,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let commit_is_versioned = opts.versioned || opts.version_suspended;
let commit_capacity_scope_token = opts.capacity_scope_token;
let commit_object_lock_guard = object_lock_guard.take();
let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some() || quota_mutation_fence;
let commit_allows_early_ack = commit_object_lock_guard.is_some();
let detach_commit_owner = commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence;
let commit = async move {
let mut _object_lock_guard = commit_object_lock_guard;
let mut _upload_guard = upload_guard;
@@ -2728,17 +2730,101 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
// `get_object_info` lookup, so the backfill has no consumer here yet.
Self::assign_rename_data_indexes(&mut parts_metadatas);
let rename_result = SetDisks::rename_data_owned(
let mut rename_result = SetDisks::rename_data_owned(
&commit_disks,
RUSTFS_META_MULTIPART_BUCKET,
&commit_upload_id_path,
(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path),
parts_metadatas,
&commit_bucket,
&commit_object,
(&commit_bucket, &commit_object),
write_quorum,
commit_allows_early_ack,
)
.await;
if quota_mutation_fence {
let mut rename_guard_release = None;
let mut needs_immediate_heal = false;
let mut tail_owns_staging_cleanup = false;
if let Ok(rename_commit) = rename_result.as_mut() {
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks);
// Install the tail watcher before any post-commit await. The
// latch keeps namespace guards through their prior handoff point.
needs_immediate_heal = rename_commit.needs_immediate_heal();
if let Some(rename_tail_drain) = rename_commit.tail_drain.take() {
tail_owns_staging_cleanup = true;
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
Some(HealChannelPriority::Normal),
Some(commit_set.pool_index),
Some(commit_set.set_index),
);
request.object_version_id = fi
.version_id
.or_else(|| commit_version_suspended.then(Uuid::nil))
.map(|version_id| version_id.to_string());
let object_lock_guard = _object_lock_guard.take();
let upload_guard = _upload_guard.take();
let cleanup_bucket = commit_bucket.clone();
let cleanup_object = commit_object.clone();
let heal_set = commit_set.clone();
let cleanup_set = commit_set.clone();
let committed_data_dir = fi.data_dir;
let cleanup_parts = parts.clone();
let cleanup_upload_path = commit_upload_id_path.clone();
let cleanup_upload_id = commit_upload_id.clone();
let fence_disks = commit_disks.clone();
let fence_tokens = quota_fence_tokens.clone();
let fence_bucket = commit_bucket.clone();
let fence_object = commit_object.clone();
let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel();
rename_guard_release = Some(guard_release_tx);
tokio::spawn(finish_rename_tail_heal(
rename_tail_drain,
guard_release_rx,
(object_lock_guard, upload_guard),
request,
move || async move {
if quota_mutation_fence {
let _ = SetDisks::release_quota_mutation_fences(
&fence_disks,
&fence_tokens,
&fence_bucket,
&fence_object,
write_quorum,
)
.await;
}
},
move |(object_lock_guard, upload_guard), targets| async move {
drop(object_lock_guard);
cleanup_set.cleanup_multipart_path(&cleanup_parts).await;
cleanup_set
.cleanup_rename_tail(
targets,
&cleanup_bucket,
&cleanup_object,
committed_data_dir,
transaction_epoch,
)
.await;
if let Err(err) = cleanup_set
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_path, write_quorum)
.await
{
warn!(
bucket = %cleanup_bucket,
object = %cleanup_object,
upload_id = %cleanup_upload_id,
error = ?err,
"completed multipart upload staging cleanup did not reach write quorum"
);
}
drop(upload_guard);
},
|request| async move { heal_set.submit_rename_tail_heal(request).await },
));
}
}
if quota_mutation_fence && !tail_owns_staging_cleanup {
let _ = SetDisks::release_quota_mutation_fences(
&commit_disks,
&quota_fence_tokens,
@@ -2755,16 +2841,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
Ok(result) => result,
Err(err) => return Err(err.into()),
};
let online_disks = rename_commit.online_disks;
let convergence = rename_commit.convergence;
let op_old_dir = rename_commit.data_dir;
let cleanup_disks = rename_commit.cleanup_disks;
let committed_file_info = rename_commit.committed_file_info;
let rename_tail_drain = rename_commit.tail_drain;
// Detach admission before any post-commit await: client cancellation
// must not couple durable convergence repair to cleanup work.
if convergence.needs_heal() {
if needs_immediate_heal {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
@@ -2802,14 +2883,15 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// 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, &commit_object) {
if let Some(release) = rename_guard_release.take() {
let _ = release.send(false);
}
return Err(StorageError::Unexpected);
}
fi = committed_file_info;
let committed_dir = fi.data_dir.unwrap_or_default().to_string();
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
fi.is_latest = true;
#[cfg(test)]
@@ -2819,30 +2901,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
.invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
.await;
if let Some(rename_tail_drain) = rename_tail_drain {
let object_lock_guard = _object_lock_guard.take();
let upload_guard = _upload_guard.take();
let tail_bucket = commit_bucket.clone();
let tail_object = commit_object.clone();
tokio::spawn(async move {
let _object_lock_guard = object_lock_guard;
let _upload_guard = upload_guard;
if let Err(err) = rename_tail_drain.await {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "failed",
bucket = %tail_bucket,
object = %tail_object,
error = %err,
"rename tail drain failed"
);
}
});
} else {
drop(_object_lock_guard.take()); // release the object lock before multipart cleanup tail IO.
if let Some(release) = rename_guard_release.take() {
let _ = release.send(true);
}
drop(_object_lock_guard.take()); // release the object lock before multipart cleanup tail IO.
#[cfg(test)]
pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterObjectPublication).await;
@@ -2855,14 +2917,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
// 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.
commit_set.cleanup_multipart_path(&parts).await;
if !tail_owns_staging_cleanup {
commit_set.cleanup_multipart_path(&parts).await;
}
if let Some(old_dir) = op_old_dir {
// backlog#898: best-effort reclaim of the dereferenced old data dir.
// Returns a receipt (never `Err`); a failed GC must not turn an
// already-committed multipart completion into a 503.
let cleanup = commit_set
.commit_rename_data_dir(
.commit_rename_data_dir_and_mark_capacity(
&cleanup_disks,
&commit_bucket,
&commit_object,
@@ -2886,9 +2950,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
#[cfg(test)]
pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
if let Err(err) = commit_set
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
.await
if !tail_owns_staging_cleanup
&& let Err(err) = commit_set
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
.await
{
warn!(
bucket = %commit_bucket,
@@ -2937,6 +3002,7 @@ mod tests {
use crate::disk::{endpoint::Endpoint, format::FormatV3};
use crate::layout::endpoints::SetupType;
use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test;
use crate::set_disk::core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier};
// No-locker helpers resolve to the isolated-context variants (see
// `hermetic_set_disks_isolated`); the guard-based tests build through
// `hermetic_set_disks_with_lockers`, which stays on the bootstrap context
@@ -2944,6 +3010,7 @@ mod tests {
use crate::set_disk::ops::object::hermetic_set_disks_support::{
hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity,
hermetic_set_disks_isolated as hermetic_set_disks, hermetic_set_disks_with_lockers,
hermetic_set_disks_with_lockers_and_ctx,
};
use crate::set_disk::ops::object::{PutObjectCommitBarrier, PutObjectCommitPause};
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
@@ -2951,7 +3018,11 @@ mod tests {
use rustfs_config::server_config::KVS;
use rustfs_lock::{LockClient, client::local::LocalClient};
use serial_test::serial;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::{
collections::HashSet,
sync::atomic::{AtomicUsize, Ordering},
time::Duration,
};
use tempfile::TempDir;
use tokio::sync::{Notify, RwLock};
@@ -3324,6 +3395,203 @@ mod tests {
)
}
#[tokio::test]
#[serial(capacity_dirty_scope)]
async fn recovered_disk_multipart_cleanup_marks_each_snapshot_it_mutates() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let recovered = disk_stores[0].clone();
let part_path = "recovered-multipart/part.1.meta";
recovered
.write_all(RUSTFS_META_MULTIPART_BUCKET, part_path, Bytes::from_static(b"stale part metadata"))
.await
.expect("recovered disk should contain staged part metadata");
set_disks.disks.write().await[0] = None;
let stale_snapshot = set_disks.get_disks_internal().await;
assert!(stale_snapshot[0].is_none(), "the pre-cleanup snapshot must exclude the disk");
set_disks.disks.write().await[0] = Some(recovered.clone());
let expected = capacity_scope_from_disks(&[Some(recovered.clone())])
.disks
.into_iter()
.collect::<HashSet<_>>();
let _ = drain_global_dirty_scopes();
set_disks.cleanup_multipart_path(&[part_path.to_string()]).await;
assert!(matches!(
recovered.read_all(RUSTFS_META_MULTIPART_BUCKET, part_path).await,
Err(DiskError::FileNotFound)
));
let part_marked = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(expected.is_subset(&part_marked), "part cleanup must mark the recovered disk it mutated");
let upload_path = "recovered-multipart/upload/part.1";
recovered
.write_all(RUSTFS_META_MULTIPART_BUCKET, upload_path, Bytes::from_static(b"stale upload shard"))
.await
.expect("recovered disk should contain staged upload data");
set_disks
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, "recovered-multipart/upload", 3)
.await
.expect("upload cleanup should use the recovered disk");
assert!(matches!(
recovered.read_all(RUSTFS_META_MULTIPART_BUCKET, upload_path).await,
Err(DiskError::FileNotFound)
));
let upload_marked = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&upload_marked),
"upload cleanup must mark the recovered disk it actually mutated"
);
}
#[tokio::test]
#[serial(capacity_dirty_scope)]
async fn early_ack_multipart_holds_quota_fences_and_re_marks_capacity_after_tail_drain() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
instance_ctx.update_erasure_type(SetupType::DistErasure).await;
let (_temp_dirs, disk_stores, set_disks) =
hermetic_set_disks_with_lockers_and_ctx(4, 0, 2, lockers, instance_ctx).await;
let bucket = "multipart-early-ack-capacity-scope";
let object = "multipart-early-ack-capacity-scope-object";
make_bucket_on_all(&disk_stores, bucket).await;
let mut initial = PutObjReader::from_vec(vec![b'0'; 1 << 20]);
set_disks
.put_object(bucket, object, &mut initial, &ObjectOptions::default())
.await
.expect("initial object should commit before the multipart overwrite");
drop(
set_disks
.acquire_write_lock_diag("multipart_capacity_fixture_ready", bucket, object)
.await
.expect("initial object tail should drain before the multipart overwrite"),
);
let multipart_body = vec![b'1'; 1 << 20];
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, &multipart_body, &ObjectOptions::default()).await;
let (staged, _) = set_disks
.check_upload_id_exists(bucket, object, &upload_id, true)
.await
.expect("staged upload metadata should be readable");
let competing_upload_id = upload_id.clone();
let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
let staged_part = format!(
"{upload_id_path}/{}/part.1",
staged.data_dir.expect("staged multipart data dir should exist")
);
signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path.clone()));
let candidate_disks = disk_stores.iter().cloned().map(Some).collect::<Vec<_>>();
let expected = capacity_scope_from_disks(&candidate_disks)
.disks
.into_iter()
.collect::<HashSet<_>>();
let _ = drain_global_dirty_scopes();
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let complete_store = Arc::clone(&set_disks);
let complete = tokio::spawn(async move {
let mut opts = ObjectOptions::default();
assert!(opts.set_quota_admission(0, u64::MAX));
complete_store
.complete_multipart_upload(bucket, object, &upload_id, parts, &opts)
.await
});
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("multipart completion should pause one tail disk during rename");
let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP);
complete
.await
.expect("early-ACK multipart task should join before tail release")
.expect("multipart completion should return after write quorum");
assert!(
rename_tasks.running() >= 1,
"the paused multipart tail disk must remain in flight after quorum ACK"
);
let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&initial),
"the multipart quorum ACK must mark every candidate disk dirty"
);
let abort_store = Arc::clone(&set_disks);
let abort = tokio::spawn(async move {
abort_store
.abort_multipart_upload(bucket, object, &competing_upload_id, &ObjectOptions::default())
.await
});
signaling.wait_for_attempts(2).await;
assert!(!abort.is_finished(), "the detached tail owner must retain the multipart upload guard");
let retained_staging = futures::future::join_all(
disk_stores
.iter()
.map(|disk| disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &staged_part)),
)
.await
.into_iter()
.filter(|result| result.is_ok())
.count();
assert_eq!(
retained_staging, 1,
"only the paused tail disk should still retain the multipart rename source"
);
signaling.set_target(rustfs_lock::ObjectKey::new(bucket, object));
let object_attempt = signaling.attempts.load(Ordering::Acquire) + 1;
let probe_store = Arc::clone(&set_disks);
let object_probe = tokio::spawn(async move {
probe_store
.acquire_write_lock_diag("multipart_tail_object_guard_probe", bucket, object)
.await
});
signaling.wait_for_attempts(object_attempt).await;
assert!(!object_probe.is_finished(), "the detached tail owner must retain the object guard");
rename_barrier.release();
object_probe
.await
.expect("object guard probe should join after the tail releases")
.expect("object guard probe should acquire after the tail releases");
tokio::time::timeout(Duration::from_secs(30), cleanup_barrier.wait_until_paused())
.await
.expect("the multipart tail should pause before reclaiming its old body");
let after_tail = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&after_tail),
"the multipart rename tail must re-mark capacity after the first scope was drained"
);
cleanup_barrier.release();
let abort_err = abort
.await
.expect("abort task should join after the tail releases")
.expect_err("the committed upload should no longer exist");
assert!(matches!(abort_err, StorageError::InvalidUploadID(..)));
for (disk_index, disk) in disk_stores.iter().enumerate() {
disk.read_version("", bucket, object, "", &ReadOptions::default())
.await
.unwrap_or_else(|err| panic!("disk {disk_index} must claim its quota fence and finish the rename: {err}"));
}
let after_cleanup = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&after_cleanup),
"the multipart tail cleanup must re-mark capacity after its preceding scope was drained"
);
})
.await;
}
async fn put_test_part(
set_disks: &Arc<SetDisks>,
bucket: &str,
@@ -7510,13 +7778,30 @@ mod tests {
let new = payload(0xC3);
let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
let parts_retry = parts_new.clone();
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
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:?}"
);
assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff");
crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
rename_barrier.release();
tokio::time::timeout(Duration::from_secs(30), async {
while rename_tasks.running() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("the crash-interrupted rename tail should drain after release");
drop(
set_disks
.acquire_write_lock_diag("post_commit_crash_tail_probe", bucket, object)
.await
.expect("the crash-interrupted tail should release its object guard"),
);
// The commit landed: the new version reads back whole and correct.
let (body, _etag) = read_object(&set_disks, bucket, object).await;
@@ -7560,7 +7845,7 @@ mod tests {
#[tokio::test]
#[serial(storage_class_env)]
async fn post_commit_crash_receipt_reclaims_old_data_after_restart() {
async fn post_commit_crash_reclaims_old_data_after_restart() {
let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "multipart-crash-old-data-receipt";
@@ -7578,30 +7863,55 @@ mod tests {
complete(&set_disks, bucket, object, &u_old, parts_old)
.await
.expect("the old version should commit");
drop(
set_disks
.acquire_write_lock_diag("post_commit_receipt_fixture_ready", bucket, object)
.await
.expect("the initial multipart tail should drain before inspecting one disk"),
);
let old_dir = current_data_dir(&disk_stores[0], bucket, object).await;
let new = payload(0x52);
let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
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 post-commit crash point must surface as unexpected, got {crashed:?}"
);
assert!(rename_tasks.running() >= 1, "the crash must interrupt an actual early-ACK tail handoff");
crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
rename_barrier.release();
tokio::time::timeout(Duration::from_secs(30), async {
while rename_tasks.running() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("the crash-interrupted rename tail should drain after release");
drop(
set_disks
.acquire_write_lock_diag("post_commit_receipt_tail_probe", bucket, object)
.await
.expect("the crash-interrupted tail should release its object guard"),
);
let (body, _) = read_object(&set_disks, bucket, object).await;
assert_eq!(body, new, "the committed replacement must remain readable after the crash");
let mut receipts = 0;
for disk in &disk_stores {
assert!(
cleanup_receipt_exists(disk, bucket, object, old_dir).await,
"post-commit crash must leave a durable old-data cleanup receipt"
);
receipts += usize::from(cleanup_receipt_exists(disk, bucket, object, old_dir).await);
assert!(
data_dir_exists(disk, bucket, object, old_dir).await,
"post-commit crash must leave old data for restart reconciliation"
);
}
assert_eq!(
receipts, 3,
"the committed quorum must persist receipts while the crash-interrupted tail preserves staging"
);
let restarted_endpoints = temp_dirs
.iter()
@@ -7646,7 +7956,12 @@ mod tests {
.reconcile_old_data_cleanup_receipts(bucket, object)
.await
.expect("restart receipt reconciliation should succeed");
assert_eq!(removed, 4, "restart reconciler should delete all receipt targets");
assert_eq!(removed, 3, "restart receipt reconciliation should delete the committed quorum's targets");
let reclaimed = restarted_set
.reclaim_orphan_data_dirs(bucket, object)
.await
.expect("restart orphan reconciliation should succeed");
assert_eq!(reclaimed, 1, "the late commit without a receipt must remain reclaimable as an orphan");
for disk in &reloaded {
assert!(
!data_dir_exists(disk, bucket, object, old_dir).await,
crates/ecstore/src/set_disk/ops/object.rs
+591 -68
View File
@@ -53,6 +53,7 @@ use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppress
use crate::object_api::{NamespaceLockFence, SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY};
use crate::services::notification_sys::RemoteVersionStateFleetProofToken;
use crate::services::tier::tier::{TierConfigMgr, TierOperationLease};
use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal};
use crate::store::ECStore;
use crate::store::utils::clean_metadata;
use futures::FutureExt as _;
@@ -2040,6 +2041,34 @@ pub(in crate::set_disk) fn merge_replication_metadata_lww(
}
impl SetDisks {
pub(in crate::set_disk) async fn cleanup_rename_tail(
&self,
targets: Vec<RenameTailCleanup>,
bucket: &str,
object: &str,
committed_data_dir: Option<Uuid>,
epoch: Option<Uuid>,
) {
for target in targets {
let mut disks = vec![None; self.set_drive_count];
disks[target.disk_index] = Some(target.disk);
self.persist_old_data_cleanup_receipts(&disks, bucket, object, target.old_data_dir, committed_data_dir, epoch)
.await;
let cleanup = self
.commit_rename_data_dir_and_mark_capacity(
&disks,
bucket,
object,
&target.old_data_dir.to_string(),
&committed_data_dir.unwrap_or_default().to_string(),
1,
)
.await;
self.report_old_data_dir_cleanup(bucket, object, &target.old_data_dir.to_string(), &cleanup)
.await;
}
}
pub(in crate::set_disk) async fn persist_old_data_cleanup_receipts(
&self,
disks: &[Option<DiskStore>],
@@ -2940,8 +2969,8 @@ impl SetDisks {
let commit_tmp_dir = tmp_dir.clone();
let commit_object_lock_guard = object_lock_guard.take();
let commit_bucket_lifecycle_guard = bucket_lifecycle_guard.take();
let detach_commit_owner =
commit_object_lock_guard.is_some() || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence;
let commit_allows_early_ack = commit_object_lock_guard.is_some();
let detach_commit_owner = commit_allows_early_ack || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence;
let commit_write_path_label = write_path.metric_label();
let commit_is_versioned = opts.versioned || opts.version_suspended;
let commit_versioned = opts.versioned;
@@ -2955,8 +2984,8 @@ impl SetDisks {
tmp_cleanup_owned = true;
let commit = move |cancellation: Option<CancellationToken>| async move {
let _object_lock_guard = commit_object_lock_guard;
let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
let mut _object_lock_guard = commit_object_lock_guard;
let mut _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
let mut quota_reservation = quota_reservation;
let rename_stage_start = Instant::now();
let pre_rename = async {
@@ -3071,20 +3100,104 @@ impl SetDisks {
}
Self::assign_rename_data_indexes(&mut parts_metadatas);
let rename_result = SetDisks::rename_data_owned_with_fence(
let mut rename_result = SetDisks::rename_data_owned_with_fence(
&commit_disks,
RUSTFS_META_TMP_BUCKET,
commit_tmp_dir.as_str(),
(RUSTFS_META_TMP_BUCKET, commit_tmp_dir.as_str()),
parts_metadatas,
&commit_bucket,
&commit_object,
(&commit_bucket, &commit_object),
commit_allows_early_ack,
crate::set_disk::core::io_primitives::RenameDataFenceOptions::new(
write_quorum,
commit_scanner_publication_lease_tokens.as_ref(),
),
)
.await;
if quota_mutation_fence {
#[cfg(any(test, feature = "test-util"))]
if rename_result.is_ok() {
pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameQuorum).await;
}
let mut rename_guard_release = None;
let mut needs_immediate_heal = false;
let mut tail_owns_tmp_cleanup = false;
if let Ok(rename_commit) = rename_result.as_mut() {
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &rename_commit.capacity_disks);
// Install the tail watcher before any post-commit await. The
// latch keeps namespace guards through their prior handoff point.
needs_immediate_heal = rename_commit.needs_immediate_heal();
if let Some(rename_tail_drain) = rename_commit.tail_drain.take() {
tail_owns_tmp_cleanup = true;
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
false,
Some(HealChannelPriority::Normal),
Some(commit_set.pool_index),
Some(commit_set.set_index),
);
request.object_version_id = committed_version_id
.or_else(|| commit_version_suspended.then(Uuid::nil))
.map(|version_id| version_id.to_string());
let object_lock_guard = _object_lock_guard.take();
let bucket_lifecycle_guard = _bucket_lifecycle_guard.take();
let cleanup_bucket = commit_bucket.clone();
let cleanup_object = commit_object.clone();
let heal_set = commit_set.clone();
let cleanup_set = commit_set.clone();
let cleanup_tmp_dir = commit_tmp_dir.clone();
let fence_disks = commit_disks.clone();
let fence_tokens = quota_fence_tokens.clone();
let fence_bucket = commit_bucket.clone();
let fence_object = commit_object.clone();
let (guard_release_tx, guard_release_rx) = tokio::sync::oneshot::channel();
rename_guard_release = Some(guard_release_tx);
tokio::spawn(finish_rename_tail_heal(
rename_tail_drain,
guard_release_rx,
(object_lock_guard, bucket_lifecycle_guard),
request,
move || async move {
if quota_mutation_fence {
let _ = SetDisks::release_quota_mutation_fences(
&fence_disks,
&fence_tokens,
&fence_bucket,
&fence_object,
write_quorum,
)
.await;
}
},
move |(object_lock_guard, bucket_lifecycle_guard), targets| async move {
drop(object_lock_guard);
drop(bucket_lifecycle_guard);
cleanup_set
.cleanup_rename_tail(
targets,
&cleanup_bucket,
&cleanup_object,
committed_data_dir,
transaction_epoch,
)
.await;
if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
tmp_dir = %cleanup_tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
},
|request| async move { heal_set.submit_rename_tail_heal(request).await },
));
}
}
#[cfg(any(test, feature = "test-util"))]
if rename_result.is_ok() {
pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::AfterRenameHandoff).await;
}
if quota_mutation_fence && !tail_owns_tmp_cleanup {
let _ = SetDisks::release_quota_mutation_fences(
&commit_disks,
&quota_fence_tokens,
@@ -3115,16 +3228,12 @@ impl SetDisks {
}
};
let online_disks = rename_commit.online_disks;
let convergence = rename_commit.convergence;
let op_old_dir = rename_commit.data_dir;
let cleanup_disks = rename_commit.cleanup_disks;
let old_current_size = rename_commit.old_current_size;
let mut fi = rename_commit.committed_file_info;
let rename_tail_drain = rename_commit.tail_drain;
// Do this before any post-commit await so request cancellation cannot
// bypass best-effort admission. A process crash before admission
// remains subject to the existing scanner reconciliation path.
if convergence.needs_heal() {
if needs_immediate_heal {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
commit_bucket.clone(),
Some(commit_object.clone()),
@@ -3133,7 +3242,9 @@ impl SetDisks {
Some(commit_set.pool_index),
Some(commit_set.set_index),
);
request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
request.object_version_id = committed_version_id
.or_else(|| commit_version_suspended.then(Uuid::nil))
.map(|version_id| version_id.to_string());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
});
@@ -3159,38 +3270,13 @@ impl SetDisks {
.invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
.await;
// `rename_data` has completed the authoritative quorum commit. With
// the default-off early-ACK experiment, tail disk rename tasks may
// still be draining after quorum. Keep the namespace guards alive
// until that drain completes so the next same-object mutation cannot
// race a background tail rename.
if let Some(rename_tail_drain) = rename_tail_drain {
let object_lock_guard = _object_lock_guard;
let bucket_lifecycle_guard = _bucket_lifecycle_guard;
let tail_bucket = commit_bucket.clone();
let tail_object = commit_object.clone();
tokio::spawn(async move {
let _object_lock_guard = object_lock_guard;
let _bucket_lifecycle_guard = bucket_lifecycle_guard;
if let Err(err) = rename_tail_drain.await {
warn!(
event = EVENT_SET_DISK_RENAME_TAIL_DRAIN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
state = "failed",
bucket = %tail_bucket,
object = %tail_object,
error = %err,
"rename tail drain failed"
);
}
});
} else {
// The exact old-data-dir reclamation below is best-effort space
// cleanup; it must not serialize the next operation on this object.
drop(_object_lock_guard);
drop(_bucket_lifecycle_guard);
if let Some(release) = rename_guard_release.take() {
let _ = release.send(true);
}
// The exact old-data-dir reclamation below is best-effort space
// cleanup; it must not serialize the next operation on this object.
drop(_object_lock_guard.take());
drop(_bucket_lifecycle_guard.take());
rustfs_io_metrics::record_put_object_stage_duration("set_disk_rename", duration_millis_f64(rename_stage_elapsed));
if (rename_stage_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
@@ -3219,7 +3305,7 @@ impl SetDisks {
// deliberately do NOT `?`-propagate it into a 503. On residue the
// report path emits the leak metric and enqueues a heal.
let cleanup = commit_set
.commit_rename_data_dir(
.commit_rename_data_dir_and_mark_capacity(
&cleanup_disks,
&commit_bucket,
&commit_object,
@@ -3255,12 +3341,9 @@ impl SetDisks {
);
}
}
if is_compressed {
record_compression_total_memory(actual_size as u64, w_size as u64).await;
}
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
fi.replication_state_internal = Some(commit_replication_state);
fi.is_latest = true;
@@ -3317,19 +3400,21 @@ impl SetDisks {
);
}
let cleanup_set = commit_set.clone();
let cleanup_tmp_dir = commit_tmp_dir.clone();
tokio::spawn(async move {
if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
tmp_dir = %cleanup_tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
});
if !tail_owns_tmp_cleanup {
let cleanup_set = commit_set.clone();
let cleanup_tmp_dir = commit_tmp_dir.clone();
tokio::spawn(async move {
if let Err(err) = cleanup_set.delete_all(RUSTFS_META_TMP_BUCKET, &cleanup_tmp_dir).await {
warn!(tmp_dir = %cleanup_tmp_dir, error = ?err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
tmp_dir = %cleanup_tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
});
}
Ok((
ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned),
@@ -4824,6 +4909,8 @@ pub enum PutObjectCommitPause {
BeforeQuotaRename,
BeforeMetadata,
BeforeTransactionEpochVerify,
AfterRenameQuorum,
AfterRenameHandoff,
}
#[cfg(any(test, feature = "test-util"))]
@@ -13986,8 +14073,10 @@ mod put_object_tmp_cleanup_tests {
use super::hermetic_set_disks_support::hermetic_set_disks_isolated as hermetic_set_disks;
use super::*;
use crate::disk::DiskAPI as _;
use crate::set_disk::core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier};
use std::time::Duration;
use crate::set_disk::core::io_primitives::{
ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier, rename_fault_injection,
};
use std::{collections::HashSet, time::Duration};
use tempfile::TempDir;
use tokio::io::AsyncReadExt;
@@ -14028,6 +14117,45 @@ mod put_object_tmp_cleanup_tests {
}
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn recovered_disk_tmp_cleanup_marks_the_snapshot_it_mutates() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let recovered = disk_stores[0].clone();
let tmp_path = "recovered-disk-cleanup/part.1";
recovered
.write_all(RUSTFS_META_TMP_BUCKET, tmp_path, Bytes::from_static(b"stale tmp shard"))
.await
.expect("recovered disk should contain staged tmp data");
set_disks.disks.write().await[0] = None;
let stale_snapshot = set_disks.get_disks_internal().await;
assert!(stale_snapshot[0].is_none(), "the pre-cleanup snapshot must exclude the disk");
set_disks.disks.write().await[0] = Some(recovered.clone());
let _ = drain_global_dirty_scopes();
set_disks
.delete_all(RUSTFS_META_TMP_BUCKET, "recovered-disk-cleanup")
.await
.expect("tmp cleanup should use the recovered disk");
assert!(matches!(
recovered.read_all(RUSTFS_META_TMP_BUCKET, tmp_path).await,
Err(DiskError::FileNotFound)
));
let expected = capacity_scope_from_disks(&[Some(recovered)])
.disks
.into_iter()
.collect::<HashSet<_>>();
let marked = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&marked),
"tmp cleanup must mark the recovered disk it actually mutated"
);
}
#[tokio::test]
async fn put_object_success_eventually_cleans_tmp_workspace() {
let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -14190,19 +14318,35 @@ mod put_object_tmp_cleanup_tests {
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn cancelled_post_commit_cleanup_does_not_retain_namespace_lock() {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async {
assert_cancelled_post_commit_cleanup_does_not_retain_namespace_lock().await;
})
.await;
}
async fn assert_cancelled_post_commit_cleanup_does_not_retain_namespace_lock() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-commit-lock-cancelled-cleanup";
let object = "commit-lock-cancelled-cleanup-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let candidate_disks = disk_stores.iter().cloned().map(Some).collect::<Vec<_>>();
let expected_scope = capacity_scope_from_disks(&candidate_disks)
.disks
.into_iter()
.collect::<HashSet<_>>();
let mut initial_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
set_disks
.put_object(bucket, object, &mut initial_reader, &ObjectOptions::default())
.await
.expect("initial object should be committed");
let _ = drain_global_dirty_scopes();
let cleanup_tasks = rename_fanout_barrier::observe_tasks(object);
let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP);
@@ -14236,6 +14380,11 @@ mod put_object_tmp_cleanup_tests {
.expect_err("the first request should be cancelled during cleanup")
.is_cancelled()
);
let committed_scope = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected_scope.is_subset(&committed_scope),
"a committed overwrite must mark every candidate disk before caller cancellation can interrupt post-commit awaits"
);
assert!(
cleanup_tasks.running() >= 1,
"cancelled cleanup must remain observable until its disk task drains"
@@ -14249,6 +14398,17 @@ mod put_object_tmp_cleanup_tests {
.await
.expect("cancelled cleanup disk tasks should drain");
drop(cleanup_barrier);
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let after_cleanup = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
if expected_scope.is_subset(&after_cleanup) {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("the completed old-data cleanup must re-mark every candidate disk after a refresh drain");
second_commit_barrier.release();
second
@@ -14263,6 +14423,7 @@ mod put_object_tmp_cleanup_tests {
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("latest body should drain");
assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
let _ = drain_global_dirty_scopes();
}
#[tokio::test]
@@ -14334,7 +14495,7 @@ mod put_object_tmp_cleanup_tests {
}
#[tokio::test]
#[serial_test::serial(rename_quorum_ack)]
#[serial_test::serial(capacity_dirty_scope)]
async fn early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes() {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -14405,6 +14566,368 @@ mod put_object_tmp_cleanup_tests {
.await;
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn early_ack_successful_tail_reclaims_its_old_non_inline_data_dir() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-early-ack-tail-cleanup";
let object = "early-ack-tail-cleanup-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let expected_scope = capacity_scope_from_disks(&disk_stores.iter().cloned().map(Some).collect::<Vec<_>>())
.disks
.into_iter()
.collect::<HashSet<_>>();
let mut initial = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
set_disks
.put_object(bucket, object, &mut initial, &ObjectOptions::default())
.await
.expect("initial non-inline object should commit");
drop(
set_disks
.acquire_write_lock_diag("tail_cleanup_fixture_ready", bucket, object)
.await
.expect("initial PUT tail should drain before inspecting one disk"),
);
let old_data_dir = disk_stores[0]
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("initial object metadata should be readable")
.data_dir
.expect("one-megabyte test object should have a data directory");
let old_part = format!("{object}/{old_data_dir}/part.1");
let mut heal_requests = set_disks.capture_test_rename_tail_heals();
let tail_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let overwrite_set = Arc::clone(&set_disks);
let overwrite = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
overwrite_set
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
});
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("one overwrite tail should pause during rename");
let cleanup_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_CLEANUP);
overwrite
.await
.expect("early-ACK overwrite task should join")
.expect("overwrite should return after write quorum");
let retained_before_tail = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part)))
.await
.into_iter()
.filter(|result| result.is_ok())
.count();
assert_eq!(
retained_before_tail, 1,
"only the paused tail disk should still retain the referenced old body"
);
let _ = drain_global_dirty_scopes();
rename_barrier.release();
tokio::time::timeout(Duration::from_secs(30), cleanup_barrier.wait_until_paused())
.await
.expect("the successful tail should pause before reclaiming its old body");
let after_tail = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected_scope.is_subset(&after_tail),
"the completed rename tail must re-mark capacity after the first scope was drained"
);
cleanup_barrier.release();
let cleanup_wait = tokio::time::timeout(Duration::from_secs(30), async {
loop {
let retained = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part)))
.await
.into_iter()
.filter(|result| result.is_ok())
.count();
if retained == 0 && tail_tasks.running() == 0 {
break;
}
tokio::task::yield_now().await;
}
})
.await;
if cleanup_wait.is_err() {
let retained = futures::future::join_all(disk_stores.iter().map(|disk| disk.read_all(bucket, &old_part)))
.await
.into_iter()
.filter(|result| result.is_ok())
.count();
panic!(
"the successful tail must reclaim its old body and drain: retained={retained}, running={}",
tail_tasks.running()
);
}
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let after_cleanup = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
if expected_scope.is_subset(&after_cleanup) {
break;
}
tokio::task::yield_now().await;
}
})
.await
.expect("the completed tail cleanup must re-mark capacity after its preceding scope was drained");
for (idx, disk) in disk_stores.iter().enumerate() {
assert!(
matches!(disk.read_all(bucket, &old_part).await, Err(DiskError::FileNotFound)),
"disk {idx} must reclaim the dereferenced old body"
);
}
let mut reader = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("the overwrite should remain readable after tail cleanup");
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("overwritten body should drain");
assert_eq!(body, vec![b'1'; TEST_OBJECT_SIZE]);
assert!(
matches!(heal_requests.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)),
"successful tail cleanup must not submit heal"
);
})
.await;
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn early_ack_put_holds_quota_fences_and_re_marks_capacity_after_tail_drain() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-early-ack-capacity-scope";
let object = "early-ack-capacity-scope-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let candidate_disks = disk_stores.iter().cloned().map(Some).collect::<Vec<_>>();
let expected = capacity_scope_from_disks(&candidate_disks)
.disks
.into_iter()
.collect::<HashSet<_>>();
let _ = drain_global_dirty_scopes();
let mut heal_requests = set_disks.capture_test_rename_tail_heals();
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let put_store = Arc::clone(&set_disks);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
let mut opts = ObjectOptions::default();
assert!(opts.set_quota_admission(0, u64::MAX));
put_store.put_object(bucket, object, &mut reader, &opts).await
});
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("PUT should pause one tail disk during rename");
put.await
.expect("early-ACK PUT task should join before tail release")
.expect("early-ACK PUT should return after write quorum");
assert!(rename_tasks.running() >= 1, "the paused tail disk must remain in flight after quorum ACK");
let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(expected.is_subset(&initial), "the quorum ACK must mark every candidate disk dirty");
rename_barrier.release();
let tail_done = tokio::time::timeout(
Duration::from_secs(30),
set_disks.acquire_write_lock_diag("capacity_scope_tail_probe", bucket, object),
)
.await
.expect("the object lock should become available after the tail drain")
.expect("the tail completion probe should acquire the object lock");
drop(tail_done);
let after_tail = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&after_tail),
"a tail completing after refresh drained the first mark must re-mark every candidate disk"
);
for (disk_index, disk) in disk_stores.iter().enumerate() {
disk.read_version("", bucket, object, "", &ReadOptions::default())
.await
.unwrap_or_else(|err| panic!("disk {disk_index} must claim its quota fence and finish the rename: {err}"));
}
assert!(
matches!(heal_requests.try_recv(), Err(tokio::sync::mpsc::error::TryRecvError::Empty)),
"a successful rename tail must not submit heal"
);
})
.await;
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn detached_early_ack_handoff_survives_caller_cancellation() {
use rustfs_object_capacity::capacity_scope::drain_global_dirty_scopes;
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-early-ack-cancelled-handoff";
let object = "early-ack-cancelled-handoff-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let candidate_disks = disk_stores.iter().cloned().map(Some).collect::<Vec<_>>();
let expected = capacity_scope_from_disks(&candidate_disks)
.disks
.into_iter()
.collect::<HashSet<_>>();
let _ = drain_global_dirty_scopes();
let mut heal_requests = set_disks.capture_test_rename_tail_heals();
let expected_version_id = Uuid::nil().to_string();
let tail_tasks = rename_fanout_barrier::observe_tasks(object);
let tail_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let _fault = rename_fault_injection::fail_rename_on(object, &[0]);
let quorum_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterRenameQuorum);
let handoff_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterRenameHandoff);
let put_store = Arc::clone(&set_disks);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
put_store
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
version_suspended: true,
..Default::default()
},
)
.await
});
tokio::time::timeout(Duration::from_secs(30), tail_barrier.wait_until_paused())
.await
.expect("one rename tail should pause while the other disks publish quorum");
quorum_barrier.wait_until_paused().await;
put.abort();
assert!(
put.await.expect_err("the request task should be cancelled").is_cancelled(),
"the caller should be cancelled after quorum publication"
);
quorum_barrier.release();
handoff_barrier.wait_until_paused().await;
assert!(tail_tasks.running() >= 1, "the detached owner must retain the paused tail");
let initial = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&initial),
"the detached continuation must install the full capacity scope"
);
assert!(
tokio::time::timeout(
Duration::from_millis(50),
set_disks.acquire_write_lock_diag("cancelled_handoff_probe", bucket, object),
)
.await
.is_err(),
"the watcher must own the namespace guard before the detached continuation awaits"
);
handoff_barrier.release();
tail_barrier.release();
let tail_done = tokio::time::timeout(
Duration::from_secs(30),
set_disks.acquire_write_lock_diag("cancelled_handoff_tail_probe", bucket, object),
)
.await
.expect("the namespace guard should release after the detached tail drains")
.expect("the post-tail probe should acquire the namespace guard");
drop(tail_done);
let after_tail = drain_global_dirty_scopes().into_iter().collect::<HashSet<_>>();
assert!(
expected.is_subset(&after_tail),
"the detached tail must re-mark capacity after the first scope was drained"
);
let request = tokio::time::timeout(Duration::from_secs(30), heal_requests.recv())
.await
.expect("a failed tail should submit heal")
.expect("the per-set heal capture should stay connected");
assert_eq!(request.bucket, bucket);
assert_eq!(request.object_prefix.as_deref(), Some(object));
assert_eq!(request.object_version_id.as_deref(), Some(expected_version_id.as_str()));
assert_eq!(request.pool_index, Some(set_disks.pool_index));
assert_eq!(request.set_index, Some(set_disks.set_index));
})
.await;
}
#[tokio::test]
#[serial_test::serial(capacity_dirty_scope)]
async fn no_lock_put_waits_for_rename_tail_under_outer_guard() {
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-no-lock-serial-rename";
let object = "put-no-lock-serial-rename-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let outer_guard = set_disks
.acquire_write_lock_diag("outer_no_lock_put", bucket, object)
.await
.expect("the outer caller should hold the namespace guard");
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let put_store = Arc::clone(&set_disks);
let put = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
put_store
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
});
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("no-lock PUT should reach the paused rename disk");
tokio::time::timeout(Duration::from_secs(30), async {
while rename_tasks.running() != 1 {
tokio::task::yield_now().await;
}
})
.await
.expect("the other rename disks should finish while one disk remains paused");
assert!(
!put.is_finished(),
"a no-lock caller that cannot hand off its outer guard must await the full rename fanout"
);
rename_barrier.release();
put.await
.expect("no-lock PUT task should join")
.expect("no-lock PUT should commit after the rename tail releases");
drop(outer_guard);
})
.await;
}
#[tokio::test]
async fn put_object_no_lock_aborts_after_outer_namespace_lock_loss() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;