gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-20 17:43:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): roll back failed delete marker writes (#7899)

Browse Source
This commit is contained in:
Chris
2026-09-15 11:44:35 +08:00
committed by GitHub
parent 6ef0740afe
commit 2fd8c9d333
3 changed files with 323 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/ops/object.rs
+2 -2
View File
@@ -7798,12 +7798,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let disk_namespace_owner = namespace_owner.clone().map(|owner| owner as Arc<dyn Send + Sync>);
rollback_futures.push(async move {
if should_rollback {
// The dedicated undo path never forwards the marker-only creation flag.
if let Err(err) = disk
.delete_version_with_namespace_owner(
.undo_write_with_namespace_owner(
&bucket,
&object,
fi,
force_del_marker,
DeleteOptions {
undo_write: true,
undo_delete: true,
crates/ecstore/src/store/object.rs
+126
View File
@@ -7812,6 +7812,132 @@ mod tests {
assert_eq!(marker_ids.len(), 2, "both delete markers must remain durable versions");
}
#[tokio::test]
async fn versioned_delete_quorum_failure_rolls_back_and_retries_require_quorum() {
for marker_copies in [0, 2, 4] {
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
let (_dirs, set_disks) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[set_disks], ctx).await);
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
let bucket = "versioned-delete-quorum-retry";
let object = "object.bin";
let opts = ObjectOptions {
versioned: true,
..Default::default()
};
let put_opts = ObjectOptions {
no_lock: true,
versioned: true,
..Default::default()
};
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created");
let set = store.pools[0].get_disks_by_key(object);
let payload = vec![0x5a; 4096];
let mut reader = PutObjReader::from_vec(payload.clone());
let original = set
.put_object(bucket, object, &mut reader, &put_opts)
.await
.expect("the versioned 4 KiB object should be written");
let disks = set.disks.read().await.clone();
// Reproduce both committed markers and residue from a failed older writer.
let marker = rustfs_filemeta::FileInfo {
name: object.to_string(),
version_id: Some(Uuid::new_v4()),
deleted: true,
mark_deleted: true,
mod_time: Some(time::OffsetDateTime::now_utc()),
..Default::default()
};
for disk in disks.iter().take(marker_copies).flatten() {
disk.delete_version(bucket, object, marker.clone(), true, crate::disk::DeleteOptions::default())
.await
.expect("the initial marker should be persisted on the selected disks");
}
let mut baseline = Vec::new();
for disk in disks.iter().flatten() {
baseline.push(
tokio::fs::read(disk.path().join(bucket).join(object).join("xl.meta"))
.await
.expect("initial metadata should exist"),
);
}
{
let mut online = set.disks.write().await;
online[2] = None;
online[3] = None;
}
for attempt in 1..=3 {
let err = store
.handle_delete_object(bucket, object, opts.clone())
.await
.expect_err("every explicit retry must fail with only two of four disks available");
assert!(
matches!(
&err,
StorageError::InsufficientWriteQuorum(error_bucket, error_object)
if error_bucket == bucket && error_object == object
),
"expected write quorum failure with {marker_copies} marker copies, attempt {attempt}: {err:?}"
);
for (index, disk) in disks.iter().enumerate() {
let metadata_path = disk
.as_ref()
.expect("saved disk should exist")
.path()
.join(bucket)
.join(object);
let actual = tokio::fs::read(metadata_path.join("xl.meta"))
.await
.expect("rollback must preserve readable metadata");
assert_eq!(
actual, baseline[index],
"failed DELETE must restore exact metadata on disk {index}, attempt {attempt}"
);
let mut entries = tokio::fs::read_dir(&metadata_path)
.await
.expect("the object directory should remain readable");
while let Some(entry) = entries.next_entry().await.expect("directory entries should be readable") {
assert!(
entry.file_name() != "xl.meta.bkp" && !entry.path().join("xl.meta.bkp").exists(),
"rollback must consume staged metadata backups on disk {index}"
);
}
}
}
*set.disks.write().await = disks;
if marker_copies != 0 {
continue;
}
let committed = store
.handle_delete_object(bucket, object, opts)
.await
.expect("DELETE should succeed after write quorum recovers");
assert!(committed.delete_marker);
assert_ne!(committed.version_id, marker.version_id);
let mut restored = Vec::new();
set.get_object_reader(
bucket,
object,
None,
HeaderMap::new(),
&ObjectOptions {
version_id: original.version_id.map(|id| id.to_string()),
..Default::default()
},
)
.await
.expect("the original version must remain readable")
.stream
.read_to_end(&mut restored)
.await
.expect("the original version should stream");
assert_eq!(restored, payload);
}
}
#[test]
fn latest_versioned_delete_marker_creation_excludes_specialized_deletes() {
assert!(latest_versioned_delete_creates_distinct_marker(&ObjectOptions {
rustfs/src/storage/rpc/node_service.rs
+195
View File
@@ -5590,6 +5590,201 @@ mod tests {
assert!(!fixture.env.ecstore.scanner_data_usage_publication_blocked().await);
}
#[cfg(not(windows))]
#[tokio::test]
async fn target_delete_quorum_failure_restores_remote_marker_and_retries_fail() {
use crate::storage::storage_api::{
contract::object::ObjectOperations,
ecstore_disk::{DeleteOptions, DiskOption, ReadOptions, new_disk},
ecstore_error::StorageError,
init_local_disks_with_instance_ctx,
};
let fixture = target_rpc_fixture().await;
let _ = rustfs_credentials::set_global_rpc_secret(Uuid::new_v4().to_string());
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind DELETE target");
let addr = listener.local_addr().expect("target address");
let set = fixture.env.ecstore.all_set_disks().into_iter().next().expect("erasure set");
let mut endpoints = fixture.env.endpoint_pools.as_ref()[0].endpoints.as_ref().clone();
let mut remote_endpoint = Endpoint::try_from(format!("http://{addr}{}", fixture.env.disk_paths[0].display()).as_str())
.expect("remote endpoint");
remote_endpoint.set_pool_index(0);
remote_endpoint.set_set_index(0);
remote_endpoint.set_disk_index(0);
remote_endpoint.is_local = true;
endpoints[0] = remote_endpoint.clone();
let mut pool = fixture.env.endpoint_pools.as_ref()[0].clone();
pool.endpoints = Endpoints::from(endpoints.clone());
init_local_disks_with_instance_ctx(&fixture.instance, EndpointServerPools::from(vec![pool]))
.await
.expect("register target endpoint using the same disk directories");
let service = make_server_for_context(Some(fixture.context.clone()));
let mut local_disks = Vec::new();
for endpoint in &endpoints {
local_disks.push(service.find_disk(&endpoint.to_string()).await.expect("registered disk"));
}
let (shutdown, stopped) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
// The production handler and protobuf transport are exercised here; authentication
// middleware has separate coverage. The client still supplies the canonical digest.
tonic::transport::Server::builder()
.add_service(NodeServiceServer::new(service))
.serve_with_incoming_shutdown(TcpListenerStream::new(listener), async {
let _ = stopped.await;
})
.await
.expect("DELETE target server");
});
remote_endpoint.is_local = false;
let remote = new_disk(
&remote_endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("remote disk client");
assert!(!remote.is_local());
let bucket = "delete-quorum-rpc";
let object = "versioned-object";
let payload = Bytes::from(vec![b'x'; 4096]);
let mut original = rustfs_filemeta::FileInfo::new(object, 1, 0);
original.erasure.index = 1;
original.version_id = Some(Uuid::new_v4());
original.mod_time = Some(OffsetDateTime::now_utc());
original.size = i64::try_from(payload.len()).expect("payload length fits metadata");
original.parts = vec![rustfs_filemeta::ObjectPartInfo {
number: 1,
size: payload.len(),
actual_size: original.size,
..Default::default()
}];
original.data = Some(payload.clone());
original.set_inline_data();
let mut before = Vec::new();
for disk in &local_disks {
disk.make_volume(bucket).await.expect("create target bucket");
disk.write_metadata(bucket, bucket, object, original.clone())
.await
.expect("seed existing version");
before.push(
tokio::fs::read(disk.path().join(bucket).join(object).join("xl.meta"))
.await
.expect("original metadata"),
);
}
let marker = rustfs_filemeta::FileInfo {
name: object.to_string(),
version_id: Some(Uuid::new_v4()),
deleted: true,
mark_deleted: true,
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
};
let rollback_dir = Uuid::new_v4();
remote
.delete_version(
bucket,
object,
marker.clone(),
true,
DeleteOptions {
old_data_dir: Some(rollback_dir),
..Default::default()
},
)
.await
.expect("forward RPC must really publish a marker");
let remote_object_dir = local_disks[0].path().join(bucket).join(object);
assert_ne!(
tokio::fs::read(remote_object_dir.join("xl.meta"))
.await
.expect("remote metadata"),
before[0]
);
assert!(remote_object_dir.join(rollback_dir.to_string()).join("xl.meta.bkp").exists());
let undo = DeleteOptions {
undo_write: true,
undo_delete: true,
old_data_dir: Some(rollback_dir),
..Default::default()
};
let rejected = remote
.delete_version(bucket, object, marker.clone(), true, undo.clone())
.await
.expect_err("forward marker flag must remain forbidden for RPC undo");
assert!(rejected.to_string().contains("undo_write cannot force a delete marker"));
remote
.delete_version(bucket, object, marker, false, undo)
.await
.expect("legal RPC undo must restore the existing version");
assert_eq!(
tokio::fs::read(remote_object_dir.join("xl.meta"))
.await
.expect("remote metadata"),
before[0]
);
assert!(!remote_object_dir.join(rollback_dir.to_string()).exists());
*set.disks.write().await = vec![Some(remote.clone()), Some(local_disks[1].clone()), None, None];
for _ in 0..3 {
let marker = rustfs_filemeta::FileInfo {
name: object.to_string(),
version_id: Some(Uuid::new_v4()),
deleted: true,
mark_deleted: true,
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
};
let result = set.delete_object_version(bucket, object, &marker, true).await;
assert!(
matches!(
&result,
Err(StorageError::InsufficientWriteQuorum(error_bucket, error_object))
if error_bucket == bucket && error_object == object
),
"{result:?}"
);
for (disk, expected) in local_disks.iter().zip(&before) {
let object_dir = disk.path().join(bucket).join(object);
assert_eq!(
tokio::fs::read(object_dir.join("xl.meta")).await.expect("restored metadata"),
*expected,
"failed quorum must restore local and remote metadata exactly"
);
let mut entries = tokio::fs::read_dir(&object_dir).await.expect("object directory");
while let Some(entry) = entries.next_entry().await.expect("object entry") {
assert_eq!(entry.file_name(), "xl.meta", "no rollback directory may remain");
}
let restored = disk
.read_version(
bucket,
bucket,
object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("latest must remain the original version");
assert_eq!(restored.version_id, original.version_id);
assert!(!restored.deleted);
assert_eq!(restored.data, Some(payload.clone()));
}
}
*set.disks.write().await = local_disks.into_iter().map(Some).collect();
remote.close().await.expect("close remote client");
let _ = shutdown.send(());
super::timeout(Duration::from_secs(10), server)
.await
.expect("server shuts down")
.expect("server task");
}
#[cfg(not(windows))]
#[tokio::test]
async fn target_handler_cancellation_retains_namespace_through_physical_rename() {