gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 06:18:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(decommission): delete marker copy waits for a busy target gate

Browse Source 
The Linux run of the S00 matrix showed the delete marker copy meeting a busy
target capacity gate without any leader kill. The copy already waits on that
gate like an ordinary object copy; pin it with an entry-level test that holds
the gate, checks the entry keeps waiting, and expects the key to migrate once
the gate is released. Share the versioned delete marker fixture with the
foreign pending slot test.
This commit is contained in:
overtrue
2026-10-20 02:49:53 +09:00
parent ab14a9b510
commit fce8626117
1 changed files with 154 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/store/init.rs
+154 -50
View File
@@ -8398,6 +8398,158 @@ mod tests {
shutdown.cancel();
}
/// Seed a versioned key with one data version and a newer delete marker on
/// the source pool, the shape whose delete marker takes the copy path.
async fn seed_decommission_delete_marker_key(
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
) -> (uuid::Uuid, uuid::Uuid) {
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create the delete marker bucket");
let data_version = uuid::Uuid::new_v4();
let marker_version = uuid::Uuid::new_v4();
let base_time = OffsetDateTime::now_utc();
seed_decommission_source(
store,
bucket,
object,
b"delete marker data".to_vec(),
&ObjectOptions {
versioned: true,
version_id: Some(data_version.to_string()),
mod_time: Some(base_time),
..Default::default()
},
)
.await;
store.pools[0]
.delete_object(
bucket,
object,
ObjectOptions {
versioned: true,
version_id: Some(marker_version.to_string()),
delete_marker: true,
mod_time: Some(base_time + time::Duration::seconds(1)),
..Default::default()
},
)
.await
.expect("seed source delete marker");
(data_version, marker_version)
}
async fn assert_decommission_delete_marker_key_migrated(
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
data_version: uuid::Uuid,
marker_version: uuid::Uuid,
) {
let marker_opts = ObjectOptions {
versioned: true,
version_id: Some(marker_version.to_string()),
..Default::default()
};
let data_opts = ObjectOptions {
versioned: true,
version_id: Some(data_version.to_string()),
..Default::default()
};
let target_marker = store.pools[1]
.get_object_info(bucket, object, &marker_opts)
.await
.expect("target delete marker must exist");
assert!(target_marker.delete_marker);
assert_eq!(
read_decommission_target_body(store, bucket, object, &data_opts).await,
b"delete marker data"
);
assert_decommission_source_absent(store, bucket, object, &marker_opts).await;
assert_decommission_source_absent(store, bucket, object, &data_opts).await;
}
/// The Linux CI run of the S00 matrix: without any kill, the delete marker
/// copy found the target capacity gate busy. It must wait for the gate like
/// an ordinary object copy and complete once the gate is released.
#[test]
#[serial_test::serial(storage_class_env)]
fn decommission_delete_marker_copy_completes_after_waiting_for_a_busy_target_gate() {
let handle = std::thread::Builder::new()
.name("decommission_delete_marker_copy_with_busy_target_gate".to_string())
.stack_size(32 * 1024 * 1024)
.spawn(|| {
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.worker_threads(2)
.build()
.expect("test runtime should build");
runtime.block_on(async {
let temp_dir = tempfile::tempdir().expect("create delete marker gate contention store dir");
let (_ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store(
temp_dir.path(),
"decommission-delete-marker-gate-contention",
&[4, 4],
))
.await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("decom-marker-gate-{}", uuid::Uuid::new_v4());
let object = "marker.bin";
let (data_version, marker_version) = seed_decommission_delete_marker_key(&store, &bucket, object).await;
mark_test_pool_decommissioning(&store, 0).await;
let incarnation = Some(store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"));
// Another writer holds the target pool's capacity gate.
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
let target_lock = store
.new_ns_lock(
crate::disk::RUSTFS_META_BUCKET,
&format!("{}/1", crate::core::pools::DECOMMISSION_CAPACITY_TARGET_LOCK_PREFIX),
)
.await
.expect("create the target capacity gate probe");
let target_guard = target_lock
.get_write_lock(std::time::Duration::from_secs(30))
.await
.expect("hold the target capacity gate");
let entry = tokio::spawn({
let store = Arc::clone(&store);
let bucket = bucket.clone();
async move {
run_decommission_entry_retry_test(
&store,
CancellationToken::new(),
&bucket,
object,
incarnation,
Arc::new(AtomicUsize::new(0)),
)
.await
}
});
tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
assert!(!entry.is_finished(), "the entry must wait while the target gate is held");
drop(target_guard);
entry
.await
.expect("the entry should join")
.expect("the delete marker copy must complete once the target gate is released");
assert_decommission_delete_marker_key_migrated(&store, &bucket, object, data_version, marker_version).await;
shutdown.cancel();
});
})
.expect("spawn delete marker gate contention test thread");
if let Err(payload) = handle.join() {
std::panic::resume_unwind(payload);
}
}
/// The S00 leader-kill run: both versions of a key were committed and
/// charged on the target, the source cleanup never ran, and on the retry a
/// concurrent mutation holds the target's pending slot between a failed
@@ -8427,40 +8579,7 @@ mod tests {
let bucket = format!("decom-marker-slot-{}", uuid::Uuid::new_v4());
let object = "marker.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create delete marker slot contention bucket");
let data_version = uuid::Uuid::new_v4();
let marker_version = uuid::Uuid::new_v4();
let base_time = OffsetDateTime::now_utc();
seed_decommission_source(
&store,
&bucket,
object,
b"delete marker data".to_vec(),
&ObjectOptions {
versioned: true,
version_id: Some(data_version.to_string()),
mod_time: Some(base_time),
..Default::default()
},
)
.await;
store.pools[0]
.delete_object(
&bucket,
object,
ObjectOptions {
versioned: true,
version_id: Some(marker_version.to_string()),
delete_marker: true,
mod_time: Some(base_time + time::Duration::seconds(1)),
..Default::default()
},
)
.await
.expect("seed source delete marker");
let (data_version, marker_version) = seed_decommission_delete_marker_key(&store, &bucket, object).await;
mark_test_pool_decommissioning(&store, 0).await;
let incarnation = Some(store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation"));
@@ -8493,11 +8612,6 @@ mod tests {
drop(barrier);
let _ = first_pass.await.expect("the first pass should join");
drop(loss_hook);
let marker_opts = ObjectOptions {
versioned: true,
version_id: Some(marker_version.to_string()),
..Default::default()
};
let data_opts = ObjectOptions {
versioned: true,
version_id: Some(data_version.to_string()),
@@ -8574,17 +8688,7 @@ mod tests {
.expect("the retried entry should join")
.expect("the entry must complete instead of failing on another mutation's pending slot");
let target_marker = store.pools[1]
.get_object_info(&bucket, object, &marker_opts)
.await
.expect("target delete marker must exist");
assert!(target_marker.delete_marker);
assert_eq!(
read_decommission_target_body(&store, &bucket, object, &data_opts).await,
b"delete marker data"
);
assert_decommission_source_absent(&store, &bucket, object, &marker_opts).await;
assert_decommission_source_absent(&store, &bucket, object, &data_opts).await;
assert_decommission_delete_marker_key_migrated(&store, &bucket, object, data_version, marker_version).await;
shutdown.cancel();
});