mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 08:38:58 +00:00
fix(ecstore): invalidate wiped disk id cache (#3251)
This commit is contained in:
@@ -60,6 +60,125 @@ mod tests {
|
||||
assert_eq!(body.as_ref(), expected, "object body changed for {key}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_auto_heal_rebuilds_runtime_wiped_disk_without_restart() {
|
||||
init_logging();
|
||||
info!("Issue #1533: auto heal should rebuild a runtime-wiped disk in a 4-disk single-node erasure set without restart");
|
||||
|
||||
let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
|
||||
let root = PathBuf::from(env.temp_dir.clone());
|
||||
let disk0 = root.join("disk0");
|
||||
let disk1 = root.join("disk1");
|
||||
let disk2 = root.join("disk2");
|
||||
let disk3 = root.join("disk3");
|
||||
for disk in [&disk0, &disk1, &disk2, &disk3] {
|
||||
std::fs::create_dir_all(disk).expect("disk directory should be created");
|
||||
}
|
||||
|
||||
env.temp_dir = disk3.to_string_lossy().to_string();
|
||||
let disk0_arg = disk0.to_string_lossy().to_string();
|
||||
let disk1_arg = disk1.to_string_lossy().to_string();
|
||||
let disk2_arg = disk2.to_string_lossy().to_string();
|
||||
env.start_rustfs_server_with_env(
|
||||
vec![disk0_arg.as_str(), disk1_arg.as_str(), disk2_arg.as_str()],
|
||||
&[
|
||||
("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true"),
|
||||
("RUSTFS_HEAL_INTERVAL_SECS", "10"),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.expect("Failed to start 4-disk RustFS");
|
||||
env.temp_dir = root.to_string_lossy().to_string();
|
||||
|
||||
let client = env.create_s3_client();
|
||||
let bucket = "heal-runtime-wiped-disk";
|
||||
let heal_timeout_secs = std::env::var("RUSTFS_AUTO_HEAL_RUNTIME_WIPE_TIMEOUT_SECS")
|
||||
.ok()
|
||||
.and_then(|value| value.parse::<u64>().ok())
|
||||
.unwrap_or(45);
|
||||
|
||||
let objects: Vec<(String, Vec<u8>, &'static str)> = vec![
|
||||
(
|
||||
"runtime/alpha.txt".to_string(),
|
||||
b"alpha payload for runtime wipe heal".to_vec(),
|
||||
"text/plain; charset=utf-8",
|
||||
),
|
||||
("runtime/beta.bin".to_string(), (0..=127).collect::<Vec<u8>>(), "application/octet-stream"),
|
||||
(
|
||||
"runtime/dir/emoji-free-name.json".to_string(),
|
||||
br#"{"status":"runtime-heal"}"#.to_vec(),
|
||||
"application/json",
|
||||
),
|
||||
(
|
||||
"runtime/dir/gamma.txt".to_string(),
|
||||
b"gamma payload for runtime wipe heal".to_vec(),
|
||||
"text/plain; charset=utf-8",
|
||||
),
|
||||
];
|
||||
|
||||
client
|
||||
.create_bucket()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("bucket create should succeed");
|
||||
for (key, body, content_type) in &objects {
|
||||
client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.content_type(*content_type)
|
||||
.body(ByteStream::from(body.clone()))
|
||||
.send()
|
||||
.await
|
||||
.expect("PUT should succeed");
|
||||
}
|
||||
|
||||
for (key, body, _) in &objects {
|
||||
assert_object_body(&env, bucket, key, body).await;
|
||||
assert!(
|
||||
object_metadata_exists_on_disk(&disk0, bucket, key),
|
||||
"disk0 should contain xl.meta before runtime wipe for {key}"
|
||||
);
|
||||
}
|
||||
|
||||
std::fs::remove_dir_all(&disk0).expect("disk0 wipe should succeed while server is running");
|
||||
std::fs::create_dir_all(&disk0).expect("disk0 should be recreated empty while server is running");
|
||||
assert!(!has_file_under(&disk0), "disk0 must be empty immediately after runtime wipe");
|
||||
|
||||
let mut remaining_rebuild_keys: HashSet<String> = objects.iter().map(|(key, _, _)| key.clone()).collect();
|
||||
for _ in 0..heal_timeout_secs {
|
||||
for (key, body, _) in &objects {
|
||||
assert_object_body(&env, bucket, key, body).await;
|
||||
}
|
||||
|
||||
if !remaining_rebuild_keys.is_empty() {
|
||||
let mut rebuilt = Vec::new();
|
||||
for key in &remaining_rebuild_keys {
|
||||
if object_metadata_exists_on_disk(&disk0, bucket, key) {
|
||||
rebuilt.push(key.clone());
|
||||
}
|
||||
}
|
||||
for key in rebuilt {
|
||||
let _ = remaining_rebuild_keys.remove(&key);
|
||||
}
|
||||
}
|
||||
|
||||
if remaining_rebuild_keys.is_empty() {
|
||||
assert!(
|
||||
disk0.join(".rustfs.sys").join("format.json").is_file(),
|
||||
"runtime-wiped disk should have format.json restored by auto heal"
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
|
||||
panic!("auto heal did not rebuild all files on the runtime-wiped disk within timeout");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_admin_deep_heal_rebuilds_cleared_disk_in_single_node_erasure_set() {
|
||||
|
||||
@@ -1793,20 +1793,28 @@ impl DiskAPI for LocalDisk {
|
||||
|
||||
let id = format_info.id;
|
||||
|
||||
// if format_info.last_check_valid() {
|
||||
// return Ok(id);
|
||||
// }
|
||||
|
||||
if format_info.file_info.is_some() && id.is_some() {
|
||||
// check last check time
|
||||
// Reuse the cached disk id only when the cached format check is fresh.
|
||||
if let Some(last_check) = format_info.last_check
|
||||
&& last_check.unix_timestamp() + 1 < OffsetDateTime::now_utc().unix_timestamp()
|
||||
&& last_check.unix_timestamp() + 1 >= OffsetDateTime::now_utc().unix_timestamp()
|
||||
{
|
||||
return Ok(id);
|
||||
}
|
||||
}
|
||||
|
||||
let file_meta = self.check_format_json().await?;
|
||||
let file_meta = match self.check_format_json().await {
|
||||
Ok(meta) => meta,
|
||||
Err(err) => {
|
||||
if matches!(err, DiskError::UnformattedDisk | DiskError::DiskNotFound) {
|
||||
let mut format_info = self.format_info.write().await;
|
||||
format_info.id = None;
|
||||
format_info.data = Bytes::new();
|
||||
format_info.file_info = None;
|
||||
format_info.last_check = None;
|
||||
}
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(file_info) = &format_info.file_info
|
||||
&& super::fs::same_file(&file_meta, file_info)
|
||||
@@ -3328,6 +3336,52 @@ mod test {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_disk_id_invalidates_cache_after_format_removal() {
|
||||
use crate::disk::FORMAT_CONFIG_FILE;
|
||||
use crate::disk::format::FormatV3;
|
||||
use tempfile::tempdir;
|
||||
|
||||
let dir = tempdir().unwrap();
|
||||
let mut endpoint = Endpoint::try_from(dir.path().to_str().unwrap()).unwrap();
|
||||
endpoint.set_pool_index(0);
|
||||
endpoint.set_set_index(0);
|
||||
endpoint.set_disk_index(0);
|
||||
let meta_dir = dir.path().join(RUSTFS_META_BUCKET);
|
||||
fs::create_dir_all(&meta_dir).await.expect("meta dir should be creatable");
|
||||
let mut format = FormatV3::new(1, 1);
|
||||
format.erasure.this = format.erasure.sets[0][0];
|
||||
let format_json = format.to_json().expect("format should serialize");
|
||||
fs::write(meta_dir.join(FORMAT_CONFIG_FILE), format_json)
|
||||
.await
|
||||
.expect("format.json should be writable");
|
||||
|
||||
let disk = LocalDisk::new(&endpoint, false)
|
||||
.await
|
||||
.expect("local disk should open after seeding format");
|
||||
|
||||
let initial_id = disk.get_disk_id().await.expect("disk id lookup should succeed");
|
||||
assert!(initial_id.is_some(), "new disk should expose a disk id");
|
||||
|
||||
fs::remove_file(&disk.format_path)
|
||||
.await
|
||||
.expect("format.json should be removable");
|
||||
|
||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||
|
||||
let err = disk
|
||||
.get_disk_id()
|
||||
.await
|
||||
.expect_err("removed format.json should invalidate the cached disk id");
|
||||
assert!(matches!(err, DiskError::UnformattedDisk));
|
||||
|
||||
let format_info = disk.format_info.read().await.clone();
|
||||
assert!(format_info.id.is_none(), "cached disk id should be cleared");
|
||||
assert!(format_info.data.is_empty(), "cached format bytes should be cleared");
|
||||
assert!(format_info.file_info.is_none(), "cached file metadata should be cleared");
|
||||
assert!(format_info.last_check.is_none(), "cached format timestamp should be cleared");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cleanup_tmp_on_startup_moves_existing_tmp_and_recreates_trash() {
|
||||
use tempfile::tempdir;
|
||||
|
||||
Reference in New Issue
Block a user