gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 19:16:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(heal,scanner): clean up dead heal/scanner code, flags, and metrics (HS-09/10/19/20) (#6256)

Browse Source
This commit is contained in:
houseme
2026-08-19 13:11:46 +08:00
committed by GitHub
parent ceb6f779fb
commit d404e1bb8a
13 changed files with 138 additions and 1049 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/channel.rs
+1 -35
View File
@@ -759,7 +759,7 @@ impl HealChannelProcessor {
#[cfg(test)]
mod tests {
use super::super::{DiskStore, Endpoint};
use super::super::DiskStore;
use super::*;
use crate::heal::manager::HealConfig;
use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
@@ -776,45 +776,18 @@ mod tests {
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> crate::Result<Option<HealObjectInfo>> {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> crate::Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> crate::Result<()> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, _object: &str) -> crate::Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> crate::Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> crate::Result<Vec<u8>> {
Ok(vec![])
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> crate::Result<crate::heal::storage::DiskStatus> {
Ok(crate::heal::storage::DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> crate::Result<()> {
Ok(())
}
async fn get_bucket_info(&self, _bucket: &str) -> crate::Result<Option<crate::heal::storage_api::status::BucketInfo>> {
Ok(None)
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> crate::Result<()> {
Ok(())
}
async fn list_buckets(&self) -> crate::Result<Vec<crate::heal::storage_api::status::BucketInfo>> {
Ok(vec![])
}
async fn object_exists(&self, _bucket: &str, _object: &str) -> crate::Result<bool> {
Ok(false)
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> crate::Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> crate::Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
_bucket: &str,
@@ -837,13 +810,6 @@ mod tests {
) -> crate::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<crate::Error>)> {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(
&self,
_bucket: &str,
_prefix: &str,
) -> crate::Result<Vec<crate::heal::storage::HealListItem>> {
Ok(vec![])
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
crates/heal/src/heal/erasure_healer.rs
+1 -31
View File
@@ -1267,7 +1267,7 @@ mod resume_loop_tests {
CheckpointManager, RESUME_CHECKPOINT_FILE, ReplacementTargetIdentity, ResumeDeleteFailure, ResumeManager, ResumeUtils,
compose_key,
};
use crate::heal::storage::{DiskStatus, HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage::{HealLifecycleExpiryContext, HealListItem, HealObjectInfo, HealStorageAPI};
use crate::heal::storage_api::status::BucketInfo;
use crate::heal::{
BUCKET_META_PREFIX, DiskOption, DiskStore, EcstoreError, Endpoint, HealDiskExt as _, RUSTFS_META_BUCKET, new_disk,
@@ -1448,36 +1448,15 @@ mod resume_loop_tests {
async fn get_object_meta(&self, _b: &str, _o: &str) -> Result<Option<HealObjectInfo>> {
Ok(None)
}
async fn get_object_data(&self, _b: &str, _o: &str) -> Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _b: &str, _o: &str, _d: &[u8]) -> Result<()> {
Ok(())
}
async fn delete_object(&self, _b: &str, _o: &str) -> Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _b: &str, _o: &str) -> Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _b: &str, _o: &str) -> Result<Vec<u8>> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _e: &Endpoint) -> Result<DiskStatus> {
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, _e: &Endpoint) -> Result<()> {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
Ok(Some(BucketInfo {
name: bucket.to_string(),
..Default::default()
}))
}
async fn heal_bucket_metadata(&self, _b: &str) -> Result<()> {
Ok(())
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
Ok(Vec::new())
}
@@ -1485,12 +1464,6 @@ mod resume_loop_tests {
// Must never be consulted: the resume loop always goes through heal_object.
panic!("object_exists must not be called by the resume heal loop");
}
async fn get_object_size(&self, _b: &str, _o: &str) -> Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _b: &str, _o: &str) -> Result<Option<String>> {
Ok(None)
}
async fn load_heal_lifecycle_expiry_context(&self, _bucket: &str) -> Result<Option<HealLifecycleExpiryContext>> {
Ok((!self.lifecycle_expired.lock().unwrap().is_empty()).then(HealLifecycleExpiryContext::test))
}
@@ -1556,9 +1529,6 @@ mod resume_loop_tests {
ReplacementCommitEvidence::Error(message) => Err(Error::other(message)),
}
}
async fn list_objects_for_heal(&self, _b: &str, _p: &str) -> Result<Vec<HealListItem>> {
Ok(Vec::new())
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
crates/heal/src/heal/manager.rs
-40
View File
@@ -3882,42 +3882,14 @@ mod tests {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<()> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> Result<Vec<u8>> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result<crate::heal::storage::DiskStatus> {
Ok(crate::heal::storage::DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> Result<()> {
Ok(())
}
async fn get_bucket_info(&self, _bucket: &str) -> Result<Option<BucketInfo>> {
Ok(None)
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> {
Ok(())
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
if let Some(hook) = manager_recovery_test_hook() {
*hook.listed.lock().expect("manager recovery listed lock should not poison") = true;
@@ -3929,14 +3901,6 @@ mod tests {
Ok(bucket == "retry-transition")
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
bucket: &str,
@@ -3998,10 +3962,6 @@ mod tests {
Ok((HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<crate::heal::storage::HealListItem>> {
Ok(Vec::new())
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
crates/heal/src/heal/storage.rs
+73 -542
View File
@@ -27,7 +27,7 @@ use super::storage_api::storage::{
BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _,
ObjectOperations as _, StorageAdminApi,
};
use super::{DiskStore, ECStore, Endpoint, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity};
use super::{DiskStore, ECStore, HealDiskExt as _, StorageError, resume::ReplacementTargetIdentity};
pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -65,7 +65,6 @@ const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_STORAGE: &str = "storage";
const EVENT_HEAL_STORAGE_OBJECT_IO: &str = "heal_storage_object_io";
const EVENT_HEAL_STORAGE_OBJECT_READ_LIMIT: &str = "heal_storage_object_read_limit";
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";
@@ -312,56 +311,23 @@ pub struct HealListItem {
pub is_delete_marker: bool,
}
/// Disk status for heal operations
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DiskStatus {
/// Ok
Ok,
/// Offline
Offline,
/// Corrupt
Corrupt,
/// Missing
Missing,
/// Permission denied
PermissionDenied,
/// Faulty
Faulty,
/// Root mount
RootMount,
/// Unknown
Unknown,
/// Unformatted
Unformatted,
}
/// Heal storage layer interface
#[async_trait]
pub trait HealStorageAPI: Send + Sync {
/// Get object meta
///
/// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): MRF intents
/// currently execute through `heal_object`; keep this entry point for the
/// metadata-corruption variant that must inspect metadata first.
async fn get_object_meta(&self, bucket: &str, object: &str) -> Result<Option<HealObjectInfo>>;
/// Get object data
async fn get_object_data(&self, bucket: &str, object: &str) -> Result<Option<Vec<u8>>>;
/// Put object data
async fn put_object_data(&self, bucket: &str, object: &str, data: &[u8]) -> Result<()>;
/// Delete object
async fn delete_object(&self, bucket: &str, object: &str) -> Result<()>;
/// Check object integrity
async fn verify_object_integrity(&self, bucket: &str, object: &str) -> Result<bool>;
/// EC decode rebuild
///
/// Reserved for HS-01 MRF wiring (rustfs/backlog#1865): urgent ECDecode
/// requests currently execute through `heal_object`; keep the explicit
/// rebuild-and-read path for the decode-failure fast variant.
async fn ec_decode_rebuild(&self, bucket: &str, object: &str) -> Result<Vec<u8>>;
/// Get disk status
async fn get_disk_status(&self, endpoint: &Endpoint) -> Result<DiskStatus>;
/// Format disk
async fn format_disk(&self, endpoint: &Endpoint) -> Result<()>;
/// Get bucket info
async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>>;
@@ -387,21 +353,12 @@ pub trait HealStorageAPI: Send + Sync {
Ok(false)
}
/// Fix bucket metadata
async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()>;
/// Get all buckets
async fn list_buckets(&self) -> Result<Vec<BucketInfo>>;
/// Check object exists
async fn object_exists(&self, bucket: &str, object: &str) -> Result<bool>;
/// Get object size
async fn get_object_size(&self, bucket: &str, object: &str) -> Result<Option<u64>>;
/// Get object checksum
async fn get_object_checksum(&self, bucket: &str, object: &str) -> Result<Option<String>>;
/// Heal object using ecstore
async fn heal_object(
&self,
@@ -453,12 +410,6 @@ pub trait HealStorageAPI: Send + Sync {
Ok(false)
}
/// List object versions for healing (returns all versions, may use significant memory for large buckets)
///
/// WARNING: This method loads all object versions into memory at once. For buckets with many
/// objects/versions, consider using `list_objects_for_heal_page` instead to process versions in pages.
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>>;
/// List object versions for healing with pagination (returns one page and continuation token)
/// Returns (versions, next_continuation_token, is_truncated). The continuation token is an
/// opaque composite `(marker, version_marker)` value — see `encode_heal_token`/`decode_heal_token`.
@@ -527,89 +478,11 @@ impl ECStoreHealStorage {
pub fn new(ecstore: Arc<ECStore>) -> Self {
Self { ecstore }
}
}
fn is_transient_object_exists_message(message: &str) -> bool {
let message = message.to_ascii_lowercase();
[
"failed to acquire read lock",
"lock acquisition failed",
"lock acquisition timeout",
"quorum not reached",
"deadline has elapsed",
"timed out",
"network error",
"transport error",
"connection refused",
]
.iter()
.any(|pattern| message.contains(pattern))
}
fn is_transient_object_exists_error(err: &StorageError) -> bool {
if err.is_quorum_error() {
return true;
}
match err {
StorageError::Lock(lock_err) => lock_err.is_retryable() || is_transient_object_exists_message(&lock_err.to_string()),
StorageError::Io(io_err) => is_transient_object_exists_message(&io_err.to_string()),
StorageError::SlowDown | StorageError::OperationCanceled => true,
_ => false,
}
}
#[async_trait]
impl HealStorageAPI for ECStoreHealStorage {
async fn get_object_meta(&self, bucket: &str, object: &str) -> Result<Option<HealObjectInfo>> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_meta",
bucket,
object,
"Heal storage request started"
);
match self.ecstore.get_object_info(bucket, object, &Default::default()).await {
Ok(info) => Ok(Some(info)),
Err(e) => {
// Map ObjectNotFound to None to align with Option return type
if matches!(e, StorageError::ObjectNotFound(_, _)) {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_meta",
bucket,
object,
result = "not_found",
"Heal storage object metadata missing"
);
Ok(None)
} else {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_meta",
bucket,
object,
result = "failed",
error = %e,
"Heal storage request failed"
);
Err(Error::other(e))
}
}
}
}
/// Read back an object's bytes, capped to bound memory.
///
/// Private support for the reserved `ec_decode_rebuild` (HS-01); not part
/// of the storage trait surface.
async fn get_object_data(&self, bucket: &str, object: &str) -> Result<Option<Vec<u8>>> {
debug!(
target: "rustfs::heal::storage",
@@ -695,196 +568,85 @@ impl HealStorageAPI for ECStoreHealStorage {
}
Ok(Some(buf))
}
}
async fn put_object_data(&self, bucket: &str, object: &str, data: &[u8]) -> Result<()> {
fn is_transient_object_exists_message(message: &str) -> bool {
let message = message.to_ascii_lowercase();
[
"failed to acquire read lock",
"lock acquisition failed",
"lock acquisition timeout",
"quorum not reached",
"deadline has elapsed",
"timed out",
"network error",
"transport error",
"connection refused",
]
.iter()
.any(|pattern| message.contains(pattern))
}
fn is_transient_object_exists_error(err: &StorageError) -> bool {
if err.is_quorum_error() {
return true;
}
match err {
StorageError::Lock(lock_err) => lock_err.is_retryable() || is_transient_object_exists_message(&lock_err.to_string()),
StorageError::Io(io_err) => is_transient_object_exists_message(&io_err.to_string()),
StorageError::SlowDown | StorageError::OperationCanceled => true,
_ => false,
}
}
#[async_trait]
impl HealStorageAPI for ECStoreHealStorage {
async fn get_object_meta(&self, bucket: &str, object: &str) -> Result<Option<HealObjectInfo>> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "put_object_data",
bucket,
object,
bytes = data.len(),
"Heal storage request started"
);
let mut reader = HealPutObjReader::from_vec(data.to_vec());
match (*self.ecstore)
.put_object(bucket, object, &mut reader, &Default::default())
.await
{
Ok(_) => {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "put_object_data",
bucket,
object,
result = "ok",
"Heal storage object write completed"
);
Ok(())
}
Err(e) => {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "put_object_data",
bucket,
object,
result = "failed",
error = %e,
"Heal storage request failed"
);
Err(Error::other(e))
}
}
}
async fn delete_object(&self, bucket: &str, object: &str) -> Result<()> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "delete_object",
operation = "get_object_meta",
bucket,
object,
"Heal storage request started"
);
match self.ecstore.delete_object(bucket, object, Default::default()).await {
Ok(_) => {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "delete_object",
bucket,
object,
result = "ok",
"Heal storage object delete completed"
);
Ok(())
}
match self.ecstore.get_object_info(bucket, object, &Default::default()).await {
Ok(info) => Ok(Some(info)),
Err(e) => {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "delete_object",
bucket,
object,
result = "failed",
error = %e,
"Heal storage request failed"
);
Err(Error::other(e))
}
}
}
async fn verify_object_integrity(&self, bucket: &str, object: &str) -> Result<bool> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
bucket,
object,
state = "started",
"Heal storage object verification started"
);
// Check object metadata first
match self.get_object_meta(bucket, object).await? {
Some(obj_info) => {
if obj_info.size < 0 {
warn!(
// Map ObjectNotFound to None to align with Option return type
if matches!(e, StorageError::ObjectNotFound(_, _)) {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_meta",
bucket,
object,
state = "invalid_size",
"Heal storage object verification failed"
result = "not_found",
"Heal storage object metadata missing"
);
return Ok(false);
Ok(None)
} else {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_meta",
bucket,
object,
result = "failed",
error = %e,
"Heal storage request failed"
);
Err(Error::other(e))
}
// Stream-read the object to a sink to avoid loading into memory
match (*self.ecstore)
.get_object_reader(bucket, object, None, Default::default(), &Default::default())
.await
{
Ok(reader) => {
let mut stream = reader.stream;
match tokio::io::copy(&mut stream, &mut tokio::io::sink()).await {
Ok(_) => {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
bucket,
object,
state = "ok",
"Heal storage object verified"
);
Ok(true)
}
Err(e) => {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
bucket,
object,
state = "stream_read_failed",
error = %e,
"Heal storage object verification failed"
);
Ok(false)
}
}
}
Err(e) => {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
bucket,
object,
state = "reader_open_failed",
error = %e,
"Heal storage object verification failed"
);
Ok(false)
}
}
}
None => {
warn!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_VERIFY,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
bucket,
object,
state = "metadata_missing",
"Heal storage object verification failed"
);
Ok(false)
}
}
}
@@ -976,81 +738,6 @@ impl HealStorageAPI for ECStoreHealStorage {
}
}
async fn get_disk_status(&self, endpoint: &Endpoint) -> Result<DiskStatus> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_disk_status",
endpoint = ?endpoint,
state = "started",
"Heal storage admin operation started"
);
// TODO: implement disk status check using ecstore
// For now, return Ok status
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_disk_status",
endpoint = ?endpoint,
result = "ok",
disk_status = "ok",
"Heal storage disk status resolved"
);
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, endpoint: &Endpoint) -> Result<()> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "format_disk",
endpoint = ?endpoint,
state = "started",
"Heal storage admin operation started"
);
// Use ecstore's heal_format
match self.heal_format(false).await {
Ok((_, error)) => {
if error.is_some() {
return Err(Error::other(format!("Format failed: {error:?}")));
}
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "format_disk",
endpoint = ?endpoint,
result = "ok",
"Heal storage disk format completed"
);
Ok(())
}
Err(e) => {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_ADMIN_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "format_disk",
endpoint = ?endpoint,
result = "failed",
error = %e,
"Heal storage admin operation failed"
);
Err(e)
}
}
}
async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
debug!(
target: "rustfs::heal::storage",
@@ -1161,61 +848,6 @@ impl HealStorageAPI for ECStoreHealStorage {
}
}
async fn heal_bucket_metadata(&self, bucket: &str) -> Result<()> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_REPAIR_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "heal_bucket_metadata",
bucket,
state = "started",
"Heal storage repair started"
);
let heal_opts = HealOpts {
recursive: true,
dry_run: false,
remove: false,
recreate: false,
scan_mode: HealScanMode::Normal,
update_parity: false,
no_lock: false,
pool: None,
set: None,
};
match self.heal_bucket(bucket, &heal_opts).await {
Ok(_) => {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_REPAIR_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "heal_bucket_metadata",
bucket,
result = "ok",
"Heal storage bucket metadata repaired"
);
Ok(())
}
Err(e) => {
error!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_REPAIR_OP,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "heal_bucket_metadata",
bucket,
result = "failed",
error = %e,
"Heal storage repair failed"
);
Err(e)
}
}
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
debug!(
target: "rustfs::heal::storage",
@@ -1315,48 +947,6 @@ impl HealStorageAPI for ECStoreHealStorage {
}
}
async fn get_object_size(&self, bucket: &str, object: &str) -> Result<Option<u64>> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_size",
bucket,
object,
"Heal storage request started"
);
match self.get_object_meta(bucket, object).await {
Ok(Some(obj_info)) => Ok(Some(obj_info.size as u64)),
Ok(None) => Ok(None),
Err(e) => Err(e),
}
}
async fn get_object_checksum(&self, bucket: &str, object: &str) -> Result<Option<String>> {
debug!(
target: "rustfs::heal::storage",
event = EVENT_HEAL_STORAGE_OBJECT_IO,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_STORAGE,
operation = "get_object_checksum",
bucket,
object,
"Heal storage request started"
);
match self.get_object_meta(bucket, object).await {
Ok(Some(obj_info)) => {
// Convert checksum bytes to hex string
let checksum = obj_info.checksum.iter().map(|b| format!("{b:02x}")).collect::<String>();
Ok(Some(checksum))
}
Ok(None) => Ok(None),
Err(e) => Err(e),
}
}
async fn heal_object(
&self,
bucket: &str,
@@ -1547,65 +1137,6 @@ impl HealStorageAPI for ECStoreHealStorage {
.map_err(Error::Storage)
}
async fn list_objects_for_heal(&self, bucket: &str, prefix: &str) -> Result<Vec<HealListItem>> {
debug!(
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 = "started",
"Heal storage admin operation started"
);
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 = "memory_heavy",
"Heal storage version listing loads all versions into memory (footprint is per-version, not per-object)"
);
let mut all_objects: Vec<HealListItem> = Vec::new();
let mut continuation_token: Option<String> = None;
loop {
let (page_objects, next_token, is_truncated) = self
.list_objects_for_heal_page(bucket, prefix, continuation_token.as_deref(), false)
.await?;
all_objects.extend(page_objects);
if !is_truncated {
break;
}
continuation_token = next_heal_listing_token(bucket, prefix, next_token, is_truncated)?;
if continuation_token.is_none() {
break;
}
}
debug!(
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,
object_count = all_objects.len(),
result = "ok",
"Heal storage object listing completed"
);
Ok(all_objects)
}
async fn list_objects_for_heal_page(
&self,
bucket: &str,
crates/heal/src/heal/task.rs
+2 -45
View File
@@ -2822,7 +2822,7 @@ impl std::fmt::Debug for HealTask {
mod tests {
use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk};
use super::*;
use crate::heal::storage::{DiskStatus, HealListItem, HealObjectInfo};
use crate::heal::storage::{HealListItem, HealObjectInfo};
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events};
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem, Infos};
use std::collections::{HashMap, VecDeque};
@@ -3354,7 +3354,6 @@ mod tests {
object_exists_by_name: Mutex<HashMap<String, MockObjectExists>>,
heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
deleted_objects: Mutex<Vec<String>>,
format_no_heal_required: Mutex<bool>,
global_format_calls: Mutex<u32>,
replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
@@ -3547,35 +3546,10 @@ mod tests {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<()> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, object: &str) -> Result<()> {
self.deleted_objects.lock().unwrap().push(object.to_string());
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> Result<Vec<u8>> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result<DiskStatus> {
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> Result<()> {
Ok(())
}
async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
Ok(Some(BucketInfo {
name: bucket.to_string(),
@@ -3590,10 +3564,6 @@ mod tests {
Ok(*self.usage_baseline.lock().unwrap())
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<()> {
Ok(())
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
let buckets = self
.listed_buckets
@@ -3621,14 +3591,6 @@ mod tests {
Ok(self.object_exists.lock().unwrap().unwrap_or(true))
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
bucket: &str,
@@ -3764,10 +3726,6 @@ mod tests {
Ok(*self.replacement_targets_ready.lock().unwrap())
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<HealListItem>> {
Ok(vec![heal_item("object-a"), heal_item("object-b")])
}
async fn list_objects_for_heal_page(
&self,
bucket: &str,
@@ -4858,7 +4816,7 @@ mod tests {
}
#[tokio::test]
async fn test_heal_failure_with_remove_corrupted_does_not_delete_object() {
async fn test_heal_failure_with_remove_corrupted_propagates_remove_flag() {
let storage = Arc::new(MockStorage {
object_exists: Mutex::new(Some(true)),
heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::OkWithOtherError(
@@ -4884,7 +4842,6 @@ mod tests {
let err = task.execute().await.expect_err("heal failure should still be reported");
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
assert!(storage.deleted_objects.lock().unwrap().is_empty());
assert!(storage.object_heal_opts.lock().unwrap()[0].remove);
}
crates/heal/src/lib.rs
+2 -42
View File
@@ -352,8 +352,8 @@ pub(crate) fn set_heal_queue_length(count: usize) {
mod tests {
use super::{
Error, HEAL_RUNTIME_INIT_TEST_HOOK, HealRuntimeInitTestHook, get_heal_channel_processor, get_heal_manager,
heal::DiskStore, heal::Endpoint, heal::manager::HealConfig, heal::storage::DiskStatus, heal::storage::HealListItem,
heal::storage::HealObjectInfo, heal::storage::HealStorageAPI, init_heal_manager, run_owned_initialization,
heal::DiskStore, heal::manager::HealConfig, heal::storage::HealListItem, heal::storage::HealObjectInfo,
heal::storage::HealStorageAPI, init_heal_manager, run_owned_initialization,
};
use crate::heal::storage_api::status::BucketInfo;
use rustfs_common::heal_channel::HealOpts;
@@ -370,42 +370,14 @@ mod tests {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> Result<Option<Vec<u8>>, Error> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> Result<(), Error> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, _object: &str) -> Result<(), Error> {
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> Result<bool, Error> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> Result<Vec<u8>, Error> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> Result<DiskStatus, Error> {
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> Result<(), Error> {
Ok(())
}
async fn get_bucket_info(&self, _bucket: &str) -> Result<Option<BucketInfo>, Error> {
Ok(None)
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> Result<(), Error> {
Ok(())
}
async fn list_buckets(&self) -> Result<Vec<BucketInfo>, Error> {
Ok(Vec::new())
}
@@ -414,14 +386,6 @@ mod tests {
Ok(false)
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> Result<Option<u64>, Error> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> Result<Option<String>, Error> {
Ok(None)
}
async fn heal_object(
&self,
_bucket: &str,
@@ -440,10 +404,6 @@ mod tests {
Ok((HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> Result<Vec<HealListItem>, Error> {
Ok(Vec::new())
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
crates/heal/tests/heal_bug_fixes_test.rs
+1 -71
View File
@@ -184,45 +184,18 @@ fn test_heal_task_status_atomic_update() {
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<HealObjectInfo>> {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Vec<u8>> {
Ok(vec![])
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<rustfs_heal::heal::storage::DiskStatus> {
Ok(rustfs_heal::heal::storage::DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<()> {
Ok(())
}
async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result<Option<BucketInfo>> {
Ok(None)
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> {
Ok(())
}
async fn list_buckets(&self) -> rustfs_heal::Result<Vec<BucketInfo>> {
Ok(vec![])
}
async fn object_exists(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<bool> {
Ok(false)
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
_bucket: &str,
@@ -245,9 +218,6 @@ fn test_heal_task_status_atomic_update() {
) -> rustfs_heal::Result<(rustfs_madmin::heal_commands::HealResultItem, Option<rustfs_heal::Error>)> {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
Ok(vec![])
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,
@@ -289,7 +259,7 @@ fn test_heal_task_status_atomic_update() {
#[tokio::test]
async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
use rustfs_heal::heal::storage::{DiskStatus, HealListItem, HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::storage::{HealListItem, HealObjectInfo, HealStorageAPI};
use rustfs_heal::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType};
use std::sync::{
Arc,
@@ -307,42 +277,14 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
Ok(None)
}
async fn get_object_data(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<Vec<u8>>> {
Ok(None)
}
async fn put_object_data(&self, _bucket: &str, _object: &str, _data: &[u8]) -> rustfs_heal::Result<()> {
Ok(())
}
async fn delete_object(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<()> {
Ok(())
}
async fn verify_object_integrity(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<bool> {
Ok(true)
}
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Vec<u8>> {
Ok(Vec::new())
}
async fn get_disk_status(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<DiskStatus> {
Ok(DiskStatus::Ok)
}
async fn format_disk(&self, _endpoint: &Endpoint) -> rustfs_heal::Result<()> {
Ok(())
}
async fn get_bucket_info(&self, _bucket: &str) -> rustfs_heal::Result<Option<BucketInfo>> {
Ok(None)
}
async fn heal_bucket_metadata(&self, _bucket: &str) -> rustfs_heal::Result<()> {
Ok(())
}
async fn list_buckets(&self) -> rustfs_heal::Result<Vec<BucketInfo>> {
Ok(Vec::new())
}
@@ -354,14 +296,6 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
))
}
async fn get_object_size(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<u64>> {
Ok(None)
}
async fn get_object_checksum(&self, _bucket: &str, _object: &str) -> rustfs_heal::Result<Option<String>> {
Ok(None)
}
async fn heal_object(
&self,
_bucket: &str,
@@ -388,10 +322,6 @@ async fn test_heal_task_transient_object_exists_skip_avoids_recreate() {
Ok((rustfs_madmin::heal_commands::HealResultItem::default(), None))
}
async fn list_objects_for_heal(&self, _bucket: &str, _prefix: &str) -> rustfs_heal::Result<Vec<HealListItem>> {
Ok(Vec::new())
}
async fn list_objects_for_heal_page(
&self,
_bucket: &str,