From 1f51d21d33adcebec4fea454807618eaf1f71c44 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 5 Jul 2026 15:53:16 +0800 Subject: [PATCH] refactor(ecstore): move List/Bucket Operations into set_disk::ops (backlog#819) (#4282) --- crates/ecstore/src/set_disk/lock.rs | 2 +- crates/ecstore/src/set_disk/mod.rs | 269 ------------------ crates/ecstore/src/set_disk/ops/bucket.rs | 229 +++++++++++++++ crates/ecstore/src/set_disk/{ => ops}/list.rs | 74 ++++- crates/ecstore/src/set_disk/ops/mod.rs | 2 + 5 files changed, 304 insertions(+), 272 deletions(-) create mode 100644 crates/ecstore/src/set_disk/ops/bucket.rs rename crates/ecstore/src/set_disk/{ => ops}/list.rs (55%) diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index 00090d767..28902f524 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -659,7 +659,7 @@ mod tests { // the List operation family must run identically through it. #[tokio::test] async fn set_disks_ctx_mirrors_core_and_drives_list_operations() { - use crate::set_disk::list::ListOperations; + use crate::set_disk::ops::list::ListOperations; let disk_count = 4; let format = FormatV3::new(1, disk_count); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 50849a3df..7f65fb29a 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -444,7 +444,6 @@ const DEFAULT_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: bool = false; static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod ctx; -mod list; mod lock; mod metadata; mod ops; @@ -3532,213 +3531,6 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { } } -#[async_trait::async_trait] -impl BucketOperations for SetDisks { - type Error = Error; - - #[tracing::instrument(skip(self))] - async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { - let disks = self.disk_inventory().await; - let write_quorum = (disks.len() / 2) + 1; - let force_create = opts.force_create; - - let mut futures = Vec::with_capacity(disks.len()); - for disk in disks { - let bucket = bucket.to_string(); - futures.push(async move { - match disk { - Some(disk) => match disk.make_volume(&bucket).await { - Ok(()) => Ok(()), - Err(err) if force_create && matches!(err, DiskError::VolumeExists) => Ok(()), - Err(err) => Err(err), - }, - None => Err(DiskError::DiskNotFound), - } - }); - } - - let results = join_all(futures).await; - let errs = results - .into_iter() - .map(|result| result.err()) - .collect::>>(); - - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { - return Err(err.into()); - } - - Ok(()) - } - - #[tracing::instrument(skip(self))] - async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result { - let disks = self.disk_inventory().await; - let write_quorum = (disks.len() / 2) + 1; - - let mut futures = Vec::with_capacity(disks.len()); - for disk in disks { - let bucket = bucket.to_string(); - futures.push(async move { - match disk { - Some(disk) => disk.stat_volume(&bucket).await, - None => Err(DiskError::DiskNotFound), - } - }); - } - - let results = join_all(futures).await; - let mut infos = Vec::with_capacity(results.len()); - let mut errs = Vec::with_capacity(results.len()); - for result in results { - match result { - Ok(info) => { - infos.push(Some(info)); - errs.push(None); - } - Err(err) => { - infos.push(None); - errs.push(Some(err)); - } - } - } - - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { - return Err(err.into()); - } - - let mut versioning = false; - let mut object_locking = false; - if let Ok(sys) = metadata_sys::get(bucket).await { - versioning = sys.versioning(); - object_locking = sys.object_locking(); - } - - infos - .into_iter() - .flatten() - .next() - .map(|info| BucketInfo { - name: info.name, - created: info.created, - versioning, - object_locking, - ..Default::default() - }) - .ok_or(Error::VolumeNotFound) - } - - #[tracing::instrument(skip(self))] - async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { - let disks = self.disk_inventory().await; - let write_quorum = (disks.len() / 2) + 1; - - let mut futures = Vec::with_capacity(disks.len()); - for disk in disks { - futures.push(async move { - match disk { - Some(disk) => disk.list_volumes().await, - None => Err(DiskError::DiskNotFound), - } - }); - } - - let results = join_all(futures).await; - let mut infos = Vec::with_capacity(results.len()); - let mut errs = Vec::with_capacity(results.len()); - for result in results { - match result { - Ok(volumes) => { - infos.push(Some(volumes)); - errs.push(None); - } - Err(err) => { - infos.push(None); - errs.push(Some(err)); - } - } - } - - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { - return Err(err.into()); - } - - let mut counts: HashMap = HashMap::new(); - for volumes in infos.into_iter().flatten() { - for volume in volumes { - if is_reserved_or_invalid_bucket(&volume.name, false) { - continue; - } - - let entry = counts.entry(volume.name.clone()).or_insert(( - 0, - BucketInfo { - name: volume.name.clone(), - created: volume.created, - ..Default::default() - }, - )); - entry.0 += 1; - } - } - - let mut buckets = counts - .into_values() - .filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket)) - .collect::>(); - buckets.sort_by(|left, right| left.name.cmp(&right.name)); - Ok(buckets) - } - - #[tracing::instrument(skip(self))] - async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { - let disks = self.disk_inventory().await; - let write_quorum = (disks.len() / 2) + 1; - - let mut futures = Vec::with_capacity(disks.len()); - for disk in disks.iter().cloned() { - let bucket = bucket.to_string(); - futures.push(async move { - match disk { - Some(disk) => disk.delete_volume(&bucket).await, - None => Err(DiskError::DiskNotFound), - } - }); - } - - let results = join_all(futures).await; - let mut errs = Vec::with_capacity(results.len()); - let mut recreate = false; - for result in results { - match result { - Ok(()) => errs.push(None), - Err(err) => { - if matches!(err, DiskError::VolumeNotEmpty) { - recreate = true; - } - errs.push(Some(err)); - } - } - } - - if recreate { - for (index, err) in errs.iter().enumerate() { - if err.is_none() - && let Some(Some(disk)) = disks.get(index) - { - let _ = disk.make_volume(bucket).await; - } - } - return Err(Error::VolumeNotEmpty); - } - - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { - return Err(err.into()); - } - - Ok(()) - } -} - fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &ObjectInfo, opts: &ObjectOptions) -> Result<()> { if let Some(retention) = &opts.object_lock_retention && check_retention_for_modification( @@ -5007,67 +4799,6 @@ impl SetDisks { } } -#[async_trait::async_trait] -impl crate::storage_api_contracts::list::ListOperations for SetDisks { - type Error = Error; - type ListObjectsV2Info = ListObjectsV2Info; - type ListObjectVersionsInfo = ListObjectVersionsInfo; - type ObjectInfoOrErr = ObjectInfoOrErr; - type WalkOptions = WalkOptions; - type WalkCancellation = CancellationToken; - type WalkResultSender = Sender; - - #[tracing::instrument(skip(self))] - async fn list_objects_v2( - self: Arc, - bucket: &str, - prefix: &str, - continuation_token: Option, - delimiter: Option, - max_keys: i32, - fetch_owner: bool, - start_after: Option, - incl_deleted: bool, - ) -> Result { - self.inner_list_objects_v2( - bucket, - prefix, - continuation_token, - delimiter, - max_keys, - fetch_owner, - start_after, - incl_deleted, - ) - .await - } - - #[tracing::instrument(skip(self))] - async fn list_object_versions( - self: Arc, - bucket: &str, - prefix: &str, - marker: Option, - version_marker: Option, - delimiter: Option, - max_keys: i32, - ) -> Result { - self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) - .await - } - - async fn walk( - self: Arc, - rx: CancellationToken, - bucket: &str, - prefix: &str, - result: Sender, - opts: WalkOptions, - ) -> Result<()> { - self.walk_internal(rx, bucket, prefix, result, opts).await - } -} - #[derive(Debug, PartialEq, Eq)] struct ObjProps { successor_mod_time: Option, diff --git a/crates/ecstore/src/set_disk/ops/bucket.rs b/crates/ecstore/src/set_disk/ops/bucket.rs new file mode 100644 index 000000000..5b3f8393b --- /dev/null +++ b/crates/ecstore/src/set_disk/ops/bucket.rs @@ -0,0 +1,229 @@ +// 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. + +//! `BucketOperations` for `SetDisks`. +//! +//! P4 of the SetDisks God-Object split (tracking backlog#815, issue #819). +//! Relocated verbatim from `set_disk/mod.rs`; the contract stays implemented +//! `for SetDisks`, so its associated-type bounds are unchanged and runtime +//! behavior is the same. + +use super::super::*; + +#[async_trait::async_trait] +impl BucketOperations for SetDisks { + type Error = Error; + + #[tracing::instrument(skip(self))] + async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { + let disks = self.disk_inventory().await; + let write_quorum = (disks.len() / 2) + 1; + let force_create = opts.force_create; + + let mut futures = Vec::with_capacity(disks.len()); + for disk in disks { + let bucket = bucket.to_string(); + futures.push(async move { + match disk { + Some(disk) => match disk.make_volume(&bucket).await { + Ok(()) => Ok(()), + Err(err) if force_create && matches!(err, DiskError::VolumeExists) => Ok(()), + Err(err) => Err(err), + }, + None => Err(DiskError::DiskNotFound), + } + }); + } + + let results = join_all(futures).await; + let errs = results + .into_iter() + .map(|result| result.err()) + .collect::>>(); + + if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { + return Err(err.into()); + } + + Ok(()) + } + + #[tracing::instrument(skip(self))] + async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result { + let disks = self.disk_inventory().await; + let write_quorum = (disks.len() / 2) + 1; + + let mut futures = Vec::with_capacity(disks.len()); + for disk in disks { + let bucket = bucket.to_string(); + futures.push(async move { + match disk { + Some(disk) => disk.stat_volume(&bucket).await, + None => Err(DiskError::DiskNotFound), + } + }); + } + + let results = join_all(futures).await; + let mut infos = Vec::with_capacity(results.len()); + let mut errs = Vec::with_capacity(results.len()); + for result in results { + match result { + Ok(info) => { + infos.push(Some(info)); + errs.push(None); + } + Err(err) => { + infos.push(None); + errs.push(Some(err)); + } + } + } + + if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { + return Err(err.into()); + } + + let mut versioning = false; + let mut object_locking = false; + if let Ok(sys) = metadata_sys::get(bucket).await { + versioning = sys.versioning(); + object_locking = sys.object_locking(); + } + + infos + .into_iter() + .flatten() + .next() + .map(|info| BucketInfo { + name: info.name, + created: info.created, + versioning, + object_locking, + ..Default::default() + }) + .ok_or(Error::VolumeNotFound) + } + + #[tracing::instrument(skip(self))] + async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { + let disks = self.disk_inventory().await; + let write_quorum = (disks.len() / 2) + 1; + + let mut futures = Vec::with_capacity(disks.len()); + for disk in disks { + futures.push(async move { + match disk { + Some(disk) => disk.list_volumes().await, + None => Err(DiskError::DiskNotFound), + } + }); + } + + let results = join_all(futures).await; + let mut infos = Vec::with_capacity(results.len()); + let mut errs = Vec::with_capacity(results.len()); + for result in results { + match result { + Ok(volumes) => { + infos.push(Some(volumes)); + errs.push(None); + } + Err(err) => { + infos.push(None); + errs.push(Some(err)); + } + } + } + + if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { + return Err(err.into()); + } + + let mut counts: HashMap = HashMap::new(); + for volumes in infos.into_iter().flatten() { + for volume in volumes { + if is_reserved_or_invalid_bucket(&volume.name, false) { + continue; + } + + let entry = counts.entry(volume.name.clone()).or_insert(( + 0, + BucketInfo { + name: volume.name.clone(), + created: volume.created, + ..Default::default() + }, + )); + entry.0 += 1; + } + } + + let mut buckets = counts + .into_values() + .filter_map(|(count, bucket)| (count >= write_quorum).then_some(bucket)) + .collect::>(); + buckets.sort_by(|left, right| left.name.cmp(&right.name)); + Ok(buckets) + } + + #[tracing::instrument(skip(self))] + async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { + let disks = self.disk_inventory().await; + let write_quorum = (disks.len() / 2) + 1; + + let mut futures = Vec::with_capacity(disks.len()); + for disk in disks.iter().cloned() { + let bucket = bucket.to_string(); + futures.push(async move { + match disk { + Some(disk) => disk.delete_volume(&bucket).await, + None => Err(DiskError::DiskNotFound), + } + }); + } + + let results = join_all(futures).await; + let mut errs = Vec::with_capacity(results.len()); + let mut recreate = false; + for result in results { + match result { + Ok(()) => errs.push(None), + Err(err) => { + if matches!(err, DiskError::VolumeNotEmpty) { + recreate = true; + } + errs.push(Some(err)); + } + } + } + + if recreate { + for (index, err) in errs.iter().enumerate() { + if err.is_none() + && let Some(Some(disk)) = disks.get(index) + { + let _ = disk.make_volume(bucket).await; + } + } + return Err(Error::VolumeNotEmpty); + } + + if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { + return Err(err.into()); + } + + Ok(()) + } +} diff --git a/crates/ecstore/src/set_disk/list.rs b/crates/ecstore/src/set_disk/ops/list.rs similarity index 55% rename from crates/ecstore/src/set_disk/list.rs rename to crates/ecstore/src/set_disk/ops/list.rs index 2cb19fb2f..11384914c 100644 --- a/crates/ecstore/src/set_disk/list.rs +++ b/crates/ecstore/src/set_disk/ops/list.rs @@ -12,8 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::ctx::SetDisksCtx; -use super::*; +//! `ListOperations` for `SetDisks` and the borrow-based prefix-maintenance +//! service unit. +//! +//! P3 of the SetDisks God-Object split (tracking backlog#815, issue #819). +//! Consolidates the `ListOperations` storage-api contract impl (relocated from +//! `set_disk/mod.rs`) with the `SetDisks::delete_all` borrow service unit +//! (relocated from `set_disk/list.rs`). Method bodies are moved verbatim and +//! runtime behavior is unchanged. + +use super::super::ctx::SetDisksCtx; +use super::super::*; impl SetDisks { #[tracing::instrument(skip(self))] @@ -91,3 +100,64 @@ impl<'a> ListOperations<'a> { Ok(()) } } + +#[async_trait::async_trait] +impl crate::storage_api_contracts::list::ListOperations for SetDisks { + type Error = Error; + type ListObjectsV2Info = ListObjectsV2Info; + type ListObjectVersionsInfo = ListObjectVersionsInfo; + type ObjectInfoOrErr = ObjectInfoOrErr; + type WalkOptions = WalkOptions; + type WalkCancellation = CancellationToken; + type WalkResultSender = Sender; + + #[tracing::instrument(skip(self))] + async fn list_objects_v2( + self: Arc, + bucket: &str, + prefix: &str, + continuation_token: Option, + delimiter: Option, + max_keys: i32, + fetch_owner: bool, + start_after: Option, + incl_deleted: bool, + ) -> Result { + self.inner_list_objects_v2( + bucket, + prefix, + continuation_token, + delimiter, + max_keys, + fetch_owner, + start_after, + incl_deleted, + ) + .await + } + + #[tracing::instrument(skip(self))] + async fn list_object_versions( + self: Arc, + bucket: &str, + prefix: &str, + marker: Option, + version_marker: Option, + delimiter: Option, + max_keys: i32, + ) -> Result { + self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) + .await + } + + async fn walk( + self: Arc, + rx: CancellationToken, + bucket: &str, + prefix: &str, + result: Sender, + opts: WalkOptions, + ) -> Result<()> { + self.walk_internal(rx, bucket, prefix, result, opts).await + } +} diff --git a/crates/ecstore/src/set_disk/ops/mod.rs b/crates/ecstore/src/set_disk/ops/mod.rs index 05b8f155e..1ea5fe940 100644 --- a/crates/ecstore/src/set_disk/ops/mod.rs +++ b/crates/ecstore/src/set_disk/ops/mod.rs @@ -17,5 +17,7 @@ //! borrows shared state through [`super::ctx::SetDisksCtx`]; the storage-api //! contract impls stay `for SetDisks`, so contract bounds are unchanged. +pub(crate) mod bucket; pub(crate) mod heal; +pub(crate) mod list; pub(crate) mod multipart;