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

refactor(ecstore): move HealOperations into set_disk::ops::heal (backlog#817) (#4267)

Browse Source 
P1 of the SetDisks split (tracking backlog#815, depends on P0 backlog#816).

Give the Heal operation family its own module home: relocate set_disk/heal.rs
to set_disk/ops/heal.rs and move the HealOperations storage-api contract impl
beside its inherent helpers. The contract stays implemented for SetDisks so its
associated-type bounds are unchanged (ecstore_contract_compat_test still
covers it).

Method bodies are moved unchanged. The four inherent helpers widen from
pub(super) to pub(in crate::set_disk) to preserve their exact prior visibility
from the deeper module. get_pool_and_set now reads topology through
SetDisksCtx to keep the Heal family aligned with the P0 borrow pattern; the
read is provably identical (ctx.format()/pool_index() alias the core fields).

Runtime behavior is unchanged.
This commit is contained in:
Zhengchao An
2026-07-05 00:53:09 +08:00
committed by GitHub
parent 6dabbaab4d
commit 55ad8df1c2
3 changed files with 203 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/mod.rs
+1 -172
View File
@@ -444,12 +444,12 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false;
static OBJECT_LOCK_DIAG_ENABLED: OnceLock<bool> = OnceLock::new();
mod ctx;
mod heal;
mod list;
mod lock;
#[path = "../metadata/set_disk.rs"]
mod metadata;
mod multipart;
mod ops;
mod read;
mod replication;
pub(crate) mod shard_source;
@@ -6082,177 +6082,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
}
}
#[async_trait::async_trait]
impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
type Error = Error;
type HealResultItem = HealResultItem;
type HealOptions = HealOpts;
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let disks = self.disks.read().await.clone();
let (formats, errs) = load_format_erasure_all(&disks, true).await;
let ref_format = match get_format_erasure_in_quorum(&formats) {
Ok(format) => format,
Err(err) => {
let can_use_cached_layout = count_errs(&errs, &DiskError::UnformattedDisk) > 0
&& formats.iter().flatten().all(|format| self.format.check_other(format).is_ok())
&& errs
.iter()
.all(|err| err.is_none() || matches!(err, Some(DiskError::UnformattedDisk)));
if can_use_cached_layout {
self.format.clone()
} else {
return Ok((HealResultItem::default(), Some(err)));
}
}
};
let endpoints = crate::layout::endpoints::Endpoints::from(self.set_endpoints.clone());
let before_drives = crate::layout::set_heal::formats_to_drives_info(&endpoints, &formats, &errs);
let mut result = HealResultItem {
heal_item_type: HealItemType::Metadata.to_string(),
detail: "disk-format".to_string(),
disk_count: self.set_drive_count,
set_count: 1,
before: Infos {
drives: before_drives.clone(),
},
after: Infos { drives: before_drives },
..Default::default()
};
if count_errs(&errs, &DiskError::UnformattedDisk) == 0 {
info!("set disk formats success, NoHealRequired, errs: {:?}", errs);
return Ok((result, Some(StorageError::NoHealRequired)));
}
if !dry_run {
for (disk_idx, err) in errs.iter().enumerate() {
if !matches!(err, Some(DiskError::UnformattedDisk)) {
continue;
}
let mut new_format = ref_format.clone();
new_format.erasure.this = ref_format.erasure.sets[self.set_index][disk_idx];
if save_format_file(&disks[disk_idx], &Some(new_format.clone())).await.is_ok() {
result.after.drives[disk_idx].uuid = new_format.erasure.this.to_string();
result.after.drives[disk_idx].state = DriveState::Ok.to_string();
}
}
}
Ok((result, None))
}
#[tracing::instrument(skip(self))]
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let mut result = heal_bucket_local_on_disks(bucket, opts, self.disk_inventory().await).await?;
result.set_count = 1;
Ok(result)
}
#[tracing::instrument(skip(self))]
async fn heal_object(
&self,
bucket: &str,
object: &str,
version_id: &str,
opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
let _write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
};
if has_suffix(object, SLASH_SEPARATOR) {
let (result, err) = self.heal_object_dir_locked(bucket, object, opts.dry_run, opts.remove).await?;
return Ok((result, err.map(|e| e.into())));
}
let disks = self.disks.read().await;
let disks = disks.clone();
let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
if DiskError::is_all_not_found(&errs) {
debug!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket,
object,
version_id,
state = "missing_object_skipped",
"Set disk heal skipped missing object"
);
let err = if !version_id.is_empty() {
Error::FileVersionNotFound
} else {
Error::FileNotFound
};
return Ok((
self.default_heal_result(FileInfo::default(), &errs, bucket, object, version_id)
.await,
Some(err),
));
}
// Heal the object.
// Pass no_lock=true since we already obtained write lock (or are already called with no_lock=true)
let mut inner_opts = *opts;
inner_opts.no_lock = true;
let (result, err) = self
.heal_object(bucket, object, version_id, &inner_opts)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
if let Some(err) = err.as_ref() {
match err {
&DiskError::FileCorrupt if opts.scan_mode != HealScanMode::Deep => {
// Instead of returning an error when a bitrot error is detected
// during a normal heal scan, heal again with bitrot flag enabled.
inner_opts.scan_mode = HealScanMode::Deep;
let (result, err) = self
.heal_object(bucket, object, version_id, &inner_opts)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
return Ok((result, err.map(|e| e.into())));
}
_ => {}
}
}
Ok((result, err.map(|e| e.into())))
}
#[tracing::instrument(skip(self))]
async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>)> {
for (set_idx, set) in self.format.erasure.sets.iter().enumerate() {
for (disk_idx, disk_id) in set.iter().enumerate() {
if disk_id.to_string() == id {
return Ok((Some(self.pool_index), Some(set_idx), Some(disk_idx)));
}
}
}
Err(Error::DiskNotFound)
}
#[tracing::instrument(skip(self))]
async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> {
// Multipart orphan reconciliation is intentionally retained above the set layer
// until there is a concrete caller and a stable lower-level contract to implement.
Err(StorageError::NotImplemented)
}
}
#[derive(Debug, PartialEq, Eq)]
struct ObjProps {
successor_mod_time: Option<OffsetDateTime>,
crates/ecstore/src/set_disk/heal.rs → crates/ecstore/src/set_disk/ops/heal.rs
+182 -5
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
use super::super::*;
use crate::io_support::bitrot::object_mmap_read_enabled;
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
@@ -22,7 +22,7 @@ const EVENT_HEAL_OBJECT_RENAME: &str = "heal_object_rename";
impl SetDisks {
#[tracing::instrument(skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))]
pub(super) async fn heal_object(
pub(in crate::set_disk) async fn heal_object(
&self,
bucket: &str,
object: &str,
@@ -644,7 +644,7 @@ impl SetDisks {
}
}
pub(super) async fn heal_object_dir_locked(
pub(in crate::set_disk) async fn heal_object_dir_locked(
&self,
bucket: &str,
object: &str,
@@ -748,7 +748,7 @@ impl SetDisks {
}
#[tracing::instrument(skip(self))]
pub(super) async fn heal_object_dir(
pub(in crate::set_disk) async fn heal_object_dir(
&self,
bucket: &str,
object: &str,
@@ -765,7 +765,7 @@ impl SetDisks {
self.heal_object_dir_locked(bucket, object, dry_run, remove).await
}
pub(super) async fn default_heal_result(
pub(in crate::set_disk) async fn default_heal_result(
&self,
lfi: FileInfo,
errs: &[Option<DiskError>],
@@ -830,3 +830,180 @@ impl SetDisks {
result
}
}
// Heal operation family: the storage-api `HealOperations` contract stays
// implemented `for SetDisks` (contract bounds unchanged) but now lives beside
// its inherent helpers in the `set_disk::ops::heal` module. Bodies are moved
// unchanged; `get_pool_and_set` reads the core through `SetDisksCtx` to keep
// the Heal family aligned with the borrow pattern from #816.
#[async_trait::async_trait]
impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
type Error = Error;
type HealResultItem = HealResultItem;
type HealOptions = HealOpts;
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let disks = self.disks.read().await.clone();
let (formats, errs) = load_format_erasure_all(&disks, true).await;
let ref_format = match get_format_erasure_in_quorum(&formats) {
Ok(format) => format,
Err(err) => {
let can_use_cached_layout = count_errs(&errs, &DiskError::UnformattedDisk) > 0
&& formats.iter().flatten().all(|format| self.format.check_other(format).is_ok())
&& errs
.iter()
.all(|err| err.is_none() || matches!(err, Some(DiskError::UnformattedDisk)));
if can_use_cached_layout {
self.format.clone()
} else {
return Ok((HealResultItem::default(), Some(err)));
}
}
};
let endpoints = crate::layout::endpoints::Endpoints::from(self.set_endpoints.clone());
let before_drives = crate::layout::set_heal::formats_to_drives_info(&endpoints, &formats, &errs);
let mut result = HealResultItem {
heal_item_type: HealItemType::Metadata.to_string(),
detail: "disk-format".to_string(),
disk_count: self.set_drive_count,
set_count: 1,
before: Infos {
drives: before_drives.clone(),
},
after: Infos { drives: before_drives },
..Default::default()
};
if count_errs(&errs, &DiskError::UnformattedDisk) == 0 {
info!("set disk formats success, NoHealRequired, errs: {:?}", errs);
return Ok((result, Some(StorageError::NoHealRequired)));
}
if !dry_run {
for (disk_idx, err) in errs.iter().enumerate() {
if !matches!(err, Some(DiskError::UnformattedDisk)) {
continue;
}
let mut new_format = ref_format.clone();
new_format.erasure.this = ref_format.erasure.sets[self.set_index][disk_idx];
if save_format_file(&disks[disk_idx], &Some(new_format.clone())).await.is_ok() {
result.after.drives[disk_idx].uuid = new_format.erasure.this.to_string();
result.after.drives[disk_idx].state = DriveState::Ok.to_string();
}
}
}
Ok((result, None))
}
#[tracing::instrument(skip(self))]
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let mut result = heal_bucket_local_on_disks(bucket, opts, self.disk_inventory().await).await?;
result.set_count = 1;
Ok(result)
}
#[tracing::instrument(skip(self))]
async fn heal_object(
&self,
bucket: &str,
object: &str,
version_id: &str,
opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
let _write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
};
if has_suffix(object, SLASH_SEPARATOR) {
let (result, err) = self.heal_object_dir_locked(bucket, object, opts.dry_run, opts.remove).await?;
return Ok((result, err.map(|e| e.into())));
}
let disks = self.disks.read().await;
let disks = disks.clone();
let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
if DiskError::is_all_not_found(&errs) {
debug!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_SET_DISK,
bucket,
object,
version_id,
state = "missing_object_skipped",
"Set disk heal skipped missing object"
);
let err = if !version_id.is_empty() {
Error::FileVersionNotFound
} else {
Error::FileNotFound
};
return Ok((
self.default_heal_result(FileInfo::default(), &errs, bucket, object, version_id)
.await,
Some(err),
));
}
// Heal the object.
// Pass no_lock=true since we already obtained write lock (or are already called with no_lock=true)
let mut inner_opts = *opts;
inner_opts.no_lock = true;
let (result, err) = self
.heal_object(bucket, object, version_id, &inner_opts)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
if let Some(err) = err.as_ref() {
match err {
&DiskError::FileCorrupt if opts.scan_mode != HealScanMode::Deep => {
// Instead of returning an error when a bitrot error is detected
// during a normal heal scan, heal again with bitrot flag enabled.
inner_opts.scan_mode = HealScanMode::Deep;
let (result, err) = self
.heal_object(bucket, object, version_id, &inner_opts)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
return Ok((result, err.map(|e| e.into())));
}
_ => {}
}
}
Ok((result, err.map(|e| e.into())))
}
#[tracing::instrument(skip(self))]
async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>)> {
let ctx = self.ctx();
for (set_idx, set) in ctx.format().erasure.sets.iter().enumerate() {
for (disk_idx, disk_id) in set.iter().enumerate() {
if disk_id.to_string() == id {
return Ok((Some(ctx.pool_index()), Some(set_idx), Some(disk_idx)));
}
}
}
Err(Error::DiskNotFound)
}
#[tracing::instrument(skip(self))]
async fn check_abandoned_parts(&self, _bucket: &str, _object: &str, _opts: &HealOpts) -> Result<()> {
// Multipart orphan reconciliation is intentionally retained above the set layer
// until there is a concrete caller and a stable lower-level contract to implement.
Err(StorageError::NotImplemented)
}
}
crates/ecstore/src/set_disk/ops/mod.rs
+20
View File
@@ -0,0 +1,20 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Home for the `SetDisks` operation families extracted during the
//! God-Object split (tracking #815). Each family owns its module here and
//! borrows shared state through [`super::ctx::SetDisksCtx`]; the storage-api
//! contract impls stay `for SetDisks`, so contract bounds are unchanged.
pub(crate) mod heal;