diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 790685cb7..fc5a8a041 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -444,12 +444,12 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false; static OBJECT_LOCK_DIAG_ENABLED: OnceLock = 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)> { - 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 { - 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)> { - 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, Option, Option)> { - 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, diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs similarity index 84% rename from crates/ecstore/src/set_disk/heal.rs rename to crates/ecstore/src/set_disk/ops/heal.rs index 98f0c6a08..b4a0d024a 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -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], @@ -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)> { + 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 { + 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)> { + 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, Option, Option)> { + 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) + } +} diff --git a/crates/ecstore/src/set_disk/ops/mod.rs b/crates/ecstore/src/set_disk/ops/mod.rs new file mode 100644 index 000000000..2dec0f60b --- /dev/null +++ b/crates/ecstore/src/set_disk/ops/mod.rs @@ -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;