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

fix(ecstore): retain commit owners across cancellation (#6068)

Browse Source
This commit is contained in:
houseme
2026-08-13 18:08:58 +08:00
committed by GitHub
parent e3c15f012c
commit 6b86d44cac
3 changed files with 501 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/core/io_primitives.rs
+6 -3
View File
@@ -2933,17 +2933,20 @@ impl SetDisks {
let results = fanout.await.map_err(|_| DiskError::Unexpected)?;
for (idx, result) in results.iter().enumerate() {
match result.as_ref().map_err(|_| DiskError::Unexpected)? {
Ok(res) => {
match result {
Ok(Ok(res)) => {
data_dirs[idx] = res.rollback_data_dir.or(res.old_data_dir);
cleanup_data_dirs[idx] = res.cleanup_data_dir;
disk_versions[idx].clone_from(&res.sign);
old_current_sizes[idx] = res.old_current_size;
errs.push(None);
}
Err(e) => {
Ok(Err(e)) => {
errs.push(Some(e.clone()));
}
Err(_) => {
errs.push(Some(DiskError::Unexpected));
}
}
}
crates/ecstore/src/set_disk/ops/multipart.rs
+208 -123
View File
@@ -2296,140 +2296,157 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
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.
let (online_disks, convergence, op_old_dir, cleanup_disks, _) = Self::rename_data(
&shuffle_disks,
RUSTFS_META_MULTIPART_BUCKET,
&upload_id_path,
&parts_metadatas,
bucket,
object,
write_quorum,
)
.await?;
// Detach admission before any post-commit await: client cancellation
// must not couple durable convergence repair to cleanup work.
if convergence.needs_heal() {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
bucket.to_string(),
Some(object.to_string()),
false,
Some(HealChannelPriority::Normal),
Some(self.pool_index),
Some(self.set_index),
);
request.object_version_id = fi
.version_id
.or_else(|| opts.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;
});
}
// 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
// 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.
// Returns a receipt (never `Err`); a failed GC must not turn an
// already-committed multipart completion into a 503.
let cleanup = self
.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
.await;
self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
.await;
}
if let Some(stage_start) = complete_tail_stage_start {
rustfs_io_metrics::record_put_object_stage_duration(
"multipart_complete_tail",
stage_start.elapsed().as_secs_f64() * 1000.0,
);
}
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
let cleanup_store = self.clone();
let cleanup_upload_id_path = upload_id_path.clone();
let cleanup_bucket = bucket.to_owned();
let cleanup_object = object.to_owned();
let cleanup_upload_id = upload_id.to_owned();
let cleanup_handle = tokio::spawn(async move {
let commit_set = self.clone();
let commit_bucket = bucket.to_owned();
let commit_object = object.to_owned();
let commit_upload_id = upload_id.to_owned();
let commit_upload_id_path = upload_id_path.clone();
let commit_version_suspended = opts.version_suspended;
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();
let commit = async move {
let _object_lock_guard = commit_object_lock_guard;
let _upload_guard = upload_guard;
if let Err(err) = cleanup_store
.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
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, &commit_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.
let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
&shuffle_disks,
RUSTFS_META_MULTIPART_BUCKET,
&commit_upload_id_path,
&parts_metadatas,
&commit_bucket,
&commit_object,
write_quorum,
)
.await?;
// Detach admission before any post-commit await: client cancellation
// must not couple durable convergence repair to cleanup work.
if convergence.needs_heal() {
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());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).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, &commit_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
// 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.
commit_set.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.
// 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(
&cleanup_disks,
&commit_bucket,
&commit_object,
&old_dir.to_string(),
&committed_dir,
write_quorum,
)
.await;
commit_set
.report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
.await;
}
if let Some(stage_start) = complete_tail_stage_start {
rustfs_io_metrics::record_put_object_stage_duration(
"multipart_complete_tail",
stage_start.elapsed().as_secs_f64() * 1000.0,
);
}
#[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
{
warn!(
bucket = %cleanup_bucket,
object = %cleanup_object,
upload_id = %cleanup_upload_id,
bucket = %commit_bucket,
object = %commit_object,
upload_id = %commit_upload_id,
error = ?err,
"completed multipart upload staging cleanup did not reach write quorum"
);
}
});
if let Err(err) = cleanup_handle.await {
warn!(
bucket = %bucket,
object = %object,
upload_id = %upload_id,
error = ?err,
"completed multipart upload staging cleanup task failed"
);
}
drop(object_lock_guard); // drop object lock guard to release the lock
for (i, op_disk) in online_disks.iter().enumerate() {
if let Some(disk) = op_disk
&& disk.is_online().await
{
fi = parts_metadatas[i].clone();
break;
for (i, op_disk) in online_disks.iter().enumerate() {
if let Some(disk) = op_disk
&& disk.is_online().await
{
fi = parts_metadatas[i].clone();
break;
}
}
commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
fi.is_latest = true;
commit_set
.invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
.await;
drop(_object_lock_guard); // drop object lock guard to release the lock
drop(_upload_guard);
Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
};
if detach_commit_owner {
tokio::spawn(commit)
.await
.map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
} else {
commit.await
}
self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
fi.is_latest = true;
self.invalidate_get_object_metadata_cache(bucket, object).await;
Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
}
}
@@ -4883,6 +4900,74 @@ mod tests {
.await;
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {
temp_env::async_with_vars(
[
(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
],
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 (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
let bucket = "multipart-cancelled-tail-lock-bucket";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let (upload_id, parts) =
stage_upload_with_create_opts(&set_disks, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
signaling.clear_observed();
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
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;
let abort_store = set_disks.clone();
let abort_upload_id = upload_id.clone();
let abort = tokio::spawn(async move {
abort_store
.abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
.await
});
signaling.wait_for_attempts(2).await;
tokio::task::yield_now().await;
assert!(!abort.is_finished(), "abort must wait while completion tail owns the upload lock");
complete.abort();
assert!(
complete
.await
.expect_err("the completion request should be cancellable while the tail is paused")
.is_cancelled()
);
tokio::task::yield_now().await;
assert!(!abort.is_finished(), "cancelling the completion waiter must not release the upload lock");
barrier.release();
let abort_err = abort
.await
.expect("abort task should not panic")
.expect_err("the committed upload should no longer exist when abort acquires the lock");
assert!(
matches!(abort_err, StorageError::InvalidUploadID(..)),
"abort should return InvalidUploadID after the detached completion tail, got {abort_err:?}"
);
},
)
.await;
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn complete_validates_parts_after_an_inflight_upload_part_commit() {
crates/ecstore/src/set_disk/ops/object.rs
+287 -150
View File
@@ -1106,6 +1106,7 @@ impl SetDisks {
let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
let mut tmp_cleanup_owned = false;
let result: Result<(ObjectInfo, Option<OldCurrentSize>)> = async {
let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
@@ -1597,169 +1598,236 @@ impl SetDisks {
});
}
let rename_stage_start = Instant::now();
let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
&shuffle_disks,
RUSTFS_META_TMP_BUCKET,
tmp_dir.as_str(),
&parts_metadatas,
bucket,
object,
write_quorum,
)
.await?;
// 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() {
let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
bucket.to_string(),
Some(object.to_string()),
false,
Some(HealChannelPriority::Normal),
Some(self.pool_index),
Some(self.set_index),
);
request.object_version_id = committed_version_id.map(|version_id| version_id.to_string());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
});
}
let commit_set = self.clone();
let commit_bucket = bucket.to_owned();
let commit_object = object.to_owned();
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();
let commit_write_path_label = write_path.metric_label();
let commit_is_versioned = opts.versioned || opts.version_suspended;
let commit_capacity_scope_token = opts.capacity_scope_token;
let commit_replication_state = replication_state_to_filemeta(&opts.put_replication_state());
tmp_cleanup_owned = true;
let rename_stage_elapsed = rename_stage_start.elapsed();
let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
self.invalidate_get_object_metadata_cache(bucket, object).await;
// `rename_data` has completed the authoritative quorum commit. 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);
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 {
warn!(
event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
stage = "rename_data",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
duration_ms = { rename_stage_ms },
let commit = async move {
let _object_lock_guard = commit_object_lock_guard;
let _bucket_lifecycle_guard = commit_bucket_lifecycle_guard;
let rename_stage_start = Instant::now();
let rename_result = SetDisks::rename_data(
&shuffle_disks,
RUSTFS_META_TMP_BUCKET,
commit_tmp_dir.as_str(),
&parts_metadatas,
&commit_bucket,
&commit_object,
write_quorum,
state = "slow",
"SetDisk commit tail stage is slow"
);
}
)
.await;
let (online_disks, convergence, op_old_dir, cleanup_disks, old_current_size) = match rename_result {
Ok(commit) => commit,
Err(err) => {
if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
} else if issue3031_diag_enabled() {
warn!(
target: "rustfs_ecstore::set_disk",
bucket = %commit_bucket,
object = %commit_object,
tmp_dir = %commit_tmp_dir,
"issue3031_put_object_tmp_cleanup_done"
);
}
return Err(err.into());
}
};
// 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() {
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.map(|version_id| version_id.to_string());
tokio::spawn(async move {
let _ = rustfs_common::heal_channel::send_heal_request(request).await;
});
}
let mut cleanup_stage_ms: Option<u64> = None;
if let Some(old_dir) = op_old_dir {
let committed_dir = committed_data_dir.unwrap_or_default().to_string();
let cleanup_stage_start = Instant::now();
// backlog#898: reclaiming the dereferenced old data dir is
// best-effort and returns a receipt (never `Err`). A failed GC
// here must not negate an already-committed, durable write, so we
// deliberately do NOT `?`-propagate it into a 503. On residue the
// report path emits the leak metric and enqueues a heal.
let cleanup = self
.commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
let rename_stage_elapsed = rename_stage_start.elapsed();
let rename_stage_ms = rename_stage_elapsed.as_millis() as u64;
commit_set
.invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
.await;
let cleanup_elapsed = cleanup_stage_start.elapsed();
let cleanup_ms = cleanup_elapsed.as_millis() as u64;
cleanup_stage_ms = Some(cleanup_ms);
rustfs_io_metrics::record_put_object_stage_duration(
"set_disk_old_data_cleanup",
duration_millis_f64(cleanup_elapsed),
);
self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
.await;
if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
// `rename_data` has completed the authoritative quorum commit. 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);
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 {
warn!(
event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
stage = "commit_rename_data_dir",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
old_dir = %old_dir,
duration_ms = cleanup_ms,
stage = "rename_data",
bucket = %commit_bucket,
object = %commit_object,
tmp_dir = %commit_tmp_dir,
duration_ms = { rename_stage_ms },
write_quorum,
state = "slow",
"SetDisk commit tail stage is slow"
);
}
let mut cleanup_stage_ms: Option<u64> = None;
if let Some(old_dir) = op_old_dir {
let committed_dir = committed_data_dir.unwrap_or_default().to_string();
let cleanup_stage_start = Instant::now();
// backlog#898: reclaiming the dereferenced old data dir is
// best-effort and returns a receipt (never `Err`). A failed GC
// here must not negate an already-committed, durable write, so we
// 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(
&cleanup_disks,
&commit_bucket,
&commit_object,
&old_dir.to_string(),
&committed_dir,
write_quorum,
)
.await;
let cleanup_elapsed = cleanup_stage_start.elapsed();
let cleanup_ms = cleanup_elapsed.as_millis() as u64;
cleanup_stage_ms = Some(cleanup_ms);
rustfs_io_metrics::record_put_object_stage_duration(
"set_disk_old_data_cleanup",
duration_millis_f64(cleanup_elapsed),
);
commit_set
.report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
.await;
if (cleanup_ms as u128) >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
warn!(
event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
stage = "commit_rename_data_dir",
bucket = %commit_bucket,
object = %commit_object,
tmp_dir = %commit_tmp_dir,
old_dir = %old_dir,
duration_ms = cleanup_ms,
write_quorum,
state = "slow",
"SetDisk commit tail stage is slow"
);
}
}
let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
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;
if issue3031_diag_enabled() {
let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
warn!(
target: "rustfs_ecstore::set_disk",
bucket = %commit_bucket,
object = %commit_object,
tmp_dir = %commit_tmp_dir,
data_dir = ?fi.data_dir,
write_quorum,
online_success_count,
op_old_dir = ?op_old_dir,
"issue3031_put_object_commit_succeeded"
);
}
let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
warn!(
event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
stage = "put_object_commit_tail",
bucket = %commit_bucket,
object = %commit_object,
tmp_dir = %commit_tmp_dir,
duration_ms = total_commit_tail_ms as u64,
write_quorum,
state = "slow",
"SetDisk commit tail is slow"
);
}
if issue3031_diag_enabled() {
warn!(
event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket = %commit_bucket,
object = %commit_object,
write_quorum,
write_path = commit_write_path_label,
writer_setup_ms,
encode_ms,
rename_ms = rename_stage_ms,
cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
cleanup_present = cleanup_stage_ms.is_some(),
commit_tail_ms = total_commit_tail_ms as u64,
result = "success",
"SetDisk put_object stage summary"
);
}
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),
old_current_size,
))
};
if detach_commit_owner {
tokio::spawn(commit)
.await
.map_err(|err| Error::other(format!("put_object commit task failed: {err}")))?
} else {
commit.await
}
let committed_metadata_slot = committed_response_metadata_slot(&online_disks, response_metadata_slot);
let mut fi = std::mem::take(&mut parts_metadatas[committed_metadata_slot]);
if is_compressed {
record_compression_total_memory(actual_size as u64, w_size as u64).await;
}
self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state()));
fi.is_latest = true;
if issue3031_diag_enabled() {
let online_success_count = online_disks.iter().filter(|disk| disk.is_some()).count();
warn!(
target: "rustfs_ecstore::set_disk",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
data_dir = ?fi.data_dir,
write_quorum,
online_success_count,
op_old_dir = ?op_old_dir,
"issue3031_put_object_commit_succeeded"
);
}
let total_commit_tail_ms = rename_stage_start.elapsed().as_millis();
if total_commit_tail_ms >= SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS {
warn!(
event = EVENT_SET_DISK_COMMIT_TAIL_SLOW,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
stage = "put_object_commit_tail",
bucket = %bucket,
object = %object,
tmp_dir = %tmp_dir,
duration_ms = total_commit_tail_ms as u64,
write_quorum,
state = "slow",
"SetDisk commit tail is slow"
);
}
if issue3031_diag_enabled() {
warn!(
event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket = %bucket,
object = %object,
write_quorum,
write_path = write_path.metric_label(),
writer_setup_ms,
encode_ms,
rename_ms = rename_stage_ms,
cleanup_ms = cleanup_stage_ms.unwrap_or_default(),
cleanup_present = cleanup_stage_ms.is_some(),
commit_tail_ms = total_commit_tail_ms as u64,
result = "success",
"SetDisk put_object stage summary"
);
}
Ok((
ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended),
old_current_size,
))
}
.await;
@@ -1795,7 +1863,8 @@ impl SetDisks {
);
}
if result.is_ok() {
if tmp_cleanup_owned && result.is_ok() {
} else if result.is_ok() {
// Success path: `rename_data` has already moved the data dir out of
// the tmp workspace and removed the (empty) tmp dir where it could,
// so this delete_all is a speculative safety net that normally hits
@@ -9762,6 +9831,74 @@ mod put_object_tmp_cleanup_tests {
assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
}
#[tokio::test]
async fn cancelled_rename_keeps_namespace_lock_until_publication() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let bucket = "put-commit-lock-cancelled-rename";
let object = "commit-lock-cancelled-rename-object";
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
let first_store = Arc::clone(&set_disks);
let first = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
first_store
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
});
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
.await
.expect("first PUT should pause during the authoritative rename");
let second_namespace_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
let second_store = Arc::clone(&set_disks);
let second = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(vec![b'2'; TEST_OBJECT_SIZE]);
second_store
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
.await
});
second_namespace_barrier.release_and_wait_until_namespace_pending().await;
first.abort();
assert!(
first
.await
.expect_err("the first request should be cancelled while rename is parked")
.is_cancelled()
);
tokio::task::yield_now().await;
assert!(
!second.is_finished(),
"the second writer must remain blocked by the cancelled commit owner"
);
rename_barrier.release();
drop(rename_barrier);
tokio::time::timeout(Duration::from_secs(30), async {
while rename_tasks.running() != 0 {
tokio::task::yield_now().await;
}
})
.await
.expect("the cancelled owner's rename fanout should drain");
second
.await
.expect("second overwrite task should join")
.expect("second overwrite should commit after the cancelled owner reaches publication");
let mut reader = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("the latest overwrite should be readable");
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]);
}
#[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;