gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-24 21:26:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden lock timeouts, health, and heal paths (#3815)

Browse Source
This commit is contained in:
cxymds
2026-06-24 15:48:06 +08:00
committed by GitHub
parent 94034ef406
commit eef8f43251
13 changed files with 400 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/erasure_healer.rs
+3 -28
View File
@@ -15,7 +15,7 @@
use crate::heal::{
progress::HealProgress,
resume::{CheckpointManager, ResumeManager, ResumeUtils},
storage::HealStorageAPI,
storage::{HealStorageAPI, next_heal_listing_token},
};
use crate::{Error, Result};
use futures::{StreamExt, future::join_all, stream::FuturesUnordered};
@@ -677,20 +677,7 @@ impl ErasureSetHealer {
break;
}
continuation_token = next_token;
if continuation_token.is_none() {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
state = "missing_continuation_token",
"Erasure set bucket listing truncated without continuation token"
);
break;
}
continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
}
Ok(())
@@ -827,19 +814,7 @@ impl ErasureSetHealer {
break;
}
continuation_token = next_token;
if continuation_token.is_none() {
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "missing_continuation_token",
"Erasure set bucket listing truncated without continuation token"
);
break;
}
continuation_token = next_heal_listing_token(bucket, "", next_token, is_truncated)?;
}
// 7. final progress update
crates/heal/src/heal/manager.rs
+30 -2
View File
@@ -487,14 +487,18 @@ fn heal_type_matches_path(heal_type: &HealType, heal_path: &str) -> bool {
HealType::Cluster => false,
HealType::Object { bucket, object, .. }
| HealType::Metadata { bucket, object }
| HealType::ECDecode { bucket, object, .. } => heal_path == bucket || heal_path == format!("{bucket}/{object}"),
| HealType::ECDecode { bucket, object, .. } => heal_path_matches_bucket_child(heal_path, bucket, object),
HealType::Bucket { bucket } => heal_path == bucket,
HealType::Prefix { bucket, prefix } => heal_path == bucket || heal_path == format!("{bucket}/{prefix}"),
HealType::Prefix { bucket, prefix } => heal_path_matches_bucket_child(heal_path, bucket, prefix),
HealType::ErasureSet { set_disk_id, .. } => heal_path == set_disk_id,
HealType::MRF { meta_path } => heal_path == meta_path.trim_matches('/'),
}
}
fn heal_path_matches_bucket_child(heal_path: &str, bucket: &str, child: &str) -> bool {
heal_path == bucket || heal_path == format!("{bucket}/{child}").trim_matches('/')
}
fn publish_active_heal_count(active_heals: &HashMap<String, Arc<HealTask>>) {
crate::set_heal_active_tasks(active_heals.len());
}
@@ -3041,6 +3045,30 @@ mod tests {
assert!(retry_request_for_result(&task, &result).is_none());
}
#[test]
fn test_heal_type_matches_path_normalizes_prefix_trailing_slash() {
let heal_type = HealType::Prefix {
bucket: "bucket".to_string(),
prefix: "logs/".to_string(),
};
assert!(heal_type_matches_path(&heal_type, "bucket"));
assert!(heal_type_matches_path(&heal_type, "bucket/logs"));
assert!(heal_type_matches_path(&heal_type, "bucket/logs/"));
}
#[test]
fn test_heal_type_matches_path_normalizes_object_trailing_slash() {
let heal_type = HealType::Object {
bucket: "bucket".to_string(),
object: "object/".to_string(),
version_id: None,
};
assert!(heal_type_matches_path(&heal_type, "bucket/object"));
assert!(heal_type_matches_path(&heal_type, "bucket/object/"));
}
async fn insert_retrying_request(manager: &HealManager, request: HealRequest) -> CancellationToken {
let task_id = request.id.clone();
let cancel_token = CancellationToken::new();
crates/heal/src/heal/storage.rs
+42 -16
View File
@@ -34,6 +34,21 @@ const EVENT_HEAL_STORAGE_OBJECT_VERIFY: &str = "heal_storage_object_verify";
const EVENT_HEAL_STORAGE_ADMIN_OP: &str = "heal_storage_admin_op";
const EVENT_HEAL_STORAGE_REPAIR_OP: &str = "heal_storage_repair_op";
pub(crate) fn next_heal_listing_token(
bucket: &str,
prefix: &str,
next_token: Option<String>,
is_truncated: bool,
) -> Result<Option<String>> {
if !is_truncated {
return Ok(None);
}
next_token.map(Some).ok_or_else(|| Error::TaskExecutionFailed {
message: format!("Object listing for {bucket}/{prefix} was truncated without continuation token"),
})
}
/// Disk status for heal operations
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DiskStatus {
@@ -1086,21 +1101,7 @@ impl HealStorageAPI for ECStoreHealStorage {
break;
}
continuation_token = next_token;
if continuation_token.is_none() {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "list_objects_for_heal",
bucket,
prefix,
state = "missing_continuation_token",
"Heal storage object listing truncated without continuation token"
);
break;
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
}
debug!(
@@ -1233,7 +1234,32 @@ impl HealStorageAPI for ECStoreHealStorage {
#[cfg(test)]
mod tests {
use super::super::StorageError;
use super::{is_transient_object_exists_error, is_transient_object_exists_message};
use super::{is_transient_object_exists_error, is_transient_object_exists_message, next_heal_listing_token};
#[test]
fn next_heal_listing_token_returns_none_for_complete_page() {
assert_eq!(
next_heal_listing_token("bucket", "prefix", None, false).expect("complete page should not fail"),
None
);
}
#[test]
fn next_heal_listing_token_returns_token_for_truncated_page() {
assert_eq!(
next_heal_listing_token("bucket", "prefix", Some("token-1".to_string()), true)
.expect("truncated page with token should continue"),
Some("token-1".to_string())
);
}
#[test]
fn next_heal_listing_token_fails_for_truncated_page_without_token() {
let err = next_heal_listing_token("bucket", "prefix", None, true).expect_err("truncated page without token must fail");
assert!(matches!(err, super::Error::TaskExecutionFailed { .. }));
assert!(err.to_string().contains("truncated without continuation token"));
}
#[test]
fn transient_object_exists_message_matches_lock_quorum_failures() {
crates/heal/src/heal/task.rs
+44 -15
View File
@@ -12,7 +12,11 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI};
use crate::heal::{
ErasureSetHealer,
progress::HealProgress,
storage::{HealStorageAPI, next_heal_listing_token},
};
use crate::{Error, Result};
use metrics::{counter, histogram};
use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
@@ -1329,20 +1333,7 @@ impl HealTask {
break;
}
continuation_token = next_token;
if continuation_token.is_none() {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket,
result = "missing_continuation_token",
"Heal bucket listing truncated without continuation token"
);
break;
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
}
if failed > 0 {
@@ -2129,6 +2120,7 @@ mod tests {
object_heal_opts: Mutex<Vec<HealOpts>>,
format_no_heal_required: Mutex<bool>,
listed_prefixes: Mutex<Vec<String>>,
truncate_without_token: Mutex<bool>,
}
#[async_trait::async_trait]
@@ -2246,6 +2238,10 @@ mod tests {
continuation_token: Option<&str>,
) -> Result<(Vec<String>, Option<String>, bool)> {
self.listed_prefixes.lock().unwrap().push(prefix.to_string());
if *self.truncate_without_token.lock().unwrap() {
return Ok((vec!["object-a".to_string()], None, true));
}
let mut listed = self.listed.lock().unwrap();
if continuation_token.is_none() && !*listed {
*listed = true;
@@ -2302,6 +2298,39 @@ mod tests {
assert_eq!(result_items.iter().filter(|item| item.object_size == 1).count(), 2);
}
#[tokio::test]
async fn test_recursive_bucket_heal_fails_when_listing_lacks_continuation_token() {
let storage = Arc::new(MockStorage {
truncate_without_token: Mutex::new(true),
..Default::default()
});
let request = HealRequest::new(
HealType::Bucket {
bucket: "bucket-a".to_string(),
},
HealOptions {
recursive: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
let err = task
.heal_bucket("bucket-a")
.await
.expect_err("recursive bucket heal must fail on incomplete pagination state");
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
assert!(err.to_string().contains("truncated without continuation token"));
assert_eq!(
storage.healed_objects.lock().unwrap().as_slice(),
["object-a".to_string()],
"the already returned page may be processed, but the task must not report success"
);
}
#[tokio::test]
async fn test_cluster_heal_visits_bucket_objects() {
let storage = Arc::new(MockStorage::default());