diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 6cd9f113f..4411e96c9 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -12,6 +12,32 @@ // See the License for the specific language governing permissions and // limitations under the License. +//! `SetDisks` — one erasure set's worth of disks and the object storage logic +//! over them. Historically a single ~19.7k-line God-Object; split into a Core +//! plus borrow-based operation families during backlog#815 (Epic #728). +//! +//! Module layout after the split: +//! +//! - `mod.rs` — the `SetDisks` core struct, its construction, and the shared +//! inherent state/helpers the operation families borrow. +//! - `ctx` — `SetDisksCtx`, the borrow context that hands operation units +//! access to the core without cloning `Arc`s (P0, backlog#816). +//! - `core::io_primitives` — the shared low-level read/write/erasure IO +//! primitives (metadata-fanout quorum, bitrot readers, rename/delete, quorum +//! helpers) that the operation families call through the core (P5, +//! backlog#820). +//! - `ops/` — one module per storage-api contract family, each +//! `impl for SetDisks`: `ops::object` (`ObjectIO` + +//! `ObjectOperations`, the object read/write hot path, P6/backlog#821), +//! `ops::heal` (`HealOperations`, P1/backlog#817), `ops::multipart` +//! (`MultipartOperations`, P2/backlog#818), `ops::list` (`ListOperations`) +//! and `ops::bucket` (`BucketOperations`, P3+P4/backlog#819), `ops::locking` +//! (`NamespaceLocking` + lock helpers, P7/backlog#822). +//! - `read.rs` — the object-read operation pipeline (`get_object_*`, +//! `read_version_optimized`) and its metadata cache, kept separate from the +//! read primitives it drives. +//! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers. + // #730: SetDisks still hosts staged read/heal/write migration helpers. #![allow(dead_code)] #![allow(unused_imports)] @@ -445,7 +471,6 @@ static OBJECT_LOCK_DIAG_ENABLED: OnceLock = OnceLock::new(); mod core; mod ctx; -mod lock; mod metadata; mod ops; mod read; @@ -2487,39 +2512,6 @@ impl SetDisks { } } -#[async_trait::async_trait] -impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { - type Error = Error; - type NamespaceLock = NamespaceLockWrapper; - - #[tracing::instrument(skip(self))] - async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { - let set_lock = if runtime_sources::setup_is_dist_erasure().await { - // Calculate quorum based on lockers count (majority) - let lockers_count = self.lockers.len(); - let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 }; - NamespaceLock::with_clients_and_quorum( - format!("set-{}-{}", self.pool_index, self.set_index), - self.lockers.clone(), - write_quorum, - ) - } else { - NamespaceLock::Local(LocalLock::new( - format!("set-{}-{}", self.pool_index, self.set_index), - self.local_lock_manager.clone(), - )) - }; - - let resource = ObjectKey { - bucket: Arc::from(bucket), - object: Arc::from(object), - version: None, - }; - - Ok(NamespaceLockWrapper::new(set_lock, resource, self.locker_owner.clone())) - } -} - 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( diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/ops/locking.rs similarity index 89% rename from crates/ecstore/src/set_disk/lock.rs rename to crates/ecstore/src/set_disk/ops/locking.rs index 28902f524..effee49fd 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/ops/locking.rs @@ -12,12 +12,52 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::*; +//! `NamespaceLocking` contract impl for `SetDisks` plus the set-disk locking +//! helpers (P7 of the God-Object split, tracking backlog#815, issue +//! backlog#822). The `NamespaceLocking` impl is relocated from set_disk/mod.rs +//! and the lock formatting/mapping helpers from set_disk/lock.rs are collected +//! here; the contract stays implemented `for SetDisks`, so its associated-type +//! bounds are unchanged and helper access is via inherent calls. + +use super::super::*; use crate::disk::health_state::DriveMembershipSnapshot; use crate::runtime::sources as runtime_sources; +#[async_trait::async_trait] +impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks { + type Error = Error; + type NamespaceLock = NamespaceLockWrapper; + + #[tracing::instrument(skip(self))] + async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { + let set_lock = if runtime_sources::setup_is_dist_erasure().await { + // Calculate quorum based on lockers count (majority) + let lockers_count = self.lockers.len(); + let write_quorum = if lockers_count > 1 { (lockers_count / 2) + 1 } else { 1 }; + NamespaceLock::with_clients_and_quorum( + format!("set-{}-{}", self.pool_index, self.set_index), + self.lockers.clone(), + write_quorum, + ) + } else { + NamespaceLock::Local(LocalLock::new( + format!("set-{}-{}", self.pool_index, self.set_index), + self.local_lock_manager.clone(), + )) + }; + + let resource = ObjectKey { + bucket: Arc::from(bucket), + object: Arc::from(object), + version: None, + }; + + Ok(NamespaceLockWrapper::new(set_lock, resource, self.locker_owner.clone())) + } +} + impl SetDisks { - pub(super) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String { + pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String { match err { LockResult::Timeout => { format!("{mode} lock acquisition timed out on {bucket}/{object} (owner={})", self.locker_owner) @@ -30,7 +70,7 @@ impl SetDisks { } } - pub(super) fn format_lock_error_from_error( + pub(in crate::set_disk) fn format_lock_error_from_error( &self, bucket: &str, object: &str, @@ -51,7 +91,7 @@ impl SetDisks { } } - pub(super) fn map_namespace_lock_error( + pub(in crate::set_disk) fn map_namespace_lock_error( &self, bucket: &str, object: &str, @@ -72,7 +112,7 @@ impl SetDisks { } } - pub(super) async fn get_disks_internal(&self) -> Vec> { + pub(in crate::set_disk) async fn get_disks_internal(&self) -> Vec> { let rl = self.disks.read().await; rl.clone() @@ -94,7 +134,7 @@ impl SetDisks { disks } - pub(super) async fn get_online_disks(&self) -> Vec> { + pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec> { let snapshot = self.drive_membership_snapshot().await; let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::>(); @@ -104,7 +144,7 @@ impl SetDisks { disks } - pub(super) async fn get_online_local_disks(&self) -> Vec> { + pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec> { let snapshot = self.drive_membership_snapshot().await; let mut disks = snapshot .strict_online_local_candidates() @@ -236,7 +276,7 @@ impl SetDisks { } } - pub(super) async fn _get_local_disks(&self) -> Vec> { + pub(in crate::set_disk) async fn _get_local_disks(&self) -> Vec> { let mut disks = self.get_disks_internal().await; let mut rng = rand::rng(); @@ -322,7 +362,7 @@ impl SetDisks { disk_lock[disk_idx] = Some(new_disk); } - pub(super) fn find_disk_index(&self, fm: &FormatV3) -> Result<(usize, usize)> { + pub(in crate::set_disk) fn find_disk_index(&self, fm: &FormatV3) -> Result<(usize, usize)> { self.format.check_other(fm)?; if fm.erasure.this.is_nil() { @@ -340,7 +380,7 @@ impl SetDisks { Err(Error::other("DriveID: not found")) } - pub(super) async fn connect_endpoint(ep: &Endpoint) -> disk::error::Result<(DiskStore, FormatV3)> { + pub(in crate::set_disk) async fn connect_endpoint(ep: &Endpoint) -> disk::error::Result<(DiskStore, FormatV3)> { let disk = new_disk( ep, &DiskOption { @@ -355,12 +395,15 @@ impl SetDisks { Ok((disk, fm)) } - pub(super) async fn get_online_disk_with_healing(&self, incl_healing: bool) -> Result<(Vec>, bool)> { + pub(in crate::set_disk) async fn get_online_disk_with_healing( + &self, + incl_healing: bool, + ) -> Result<(Vec>, bool)> { let (new_disks, _, healing) = self.get_online_disk_with_healing_and_info(incl_healing).await?; Ok((new_disks, healing > 0)) } - pub(super) async fn get_online_disk_with_healing_and_info( + pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info( &self, incl_healing: bool, ) -> Result<(Vec>, Vec, usize)> { diff --git a/crates/ecstore/src/set_disk/ops/mod.rs b/crates/ecstore/src/set_disk/ops/mod.rs index f475236c0..9b45188ef 100644 --- a/crates/ecstore/src/set_disk/ops/mod.rs +++ b/crates/ecstore/src/set_disk/ops/mod.rs @@ -20,5 +20,6 @@ pub(crate) mod bucket; pub(crate) mod heal; pub(crate) mod list; +pub(crate) mod locking; pub(crate) mod multipart; pub(crate) mod object; diff --git a/crates/ecstore/src/storage_api_contracts/mod.rs b/crates/ecstore/src/storage_api_contracts/mod.rs index 32f591ef0..de8986cb4 100644 --- a/crates/ecstore/src/storage_api_contracts/mod.rs +++ b/crates/ecstore/src/storage_api_contracts/mod.rs @@ -1,6 +1,3 @@ -// #730: storage contract traits are staged for facade migration and external owners. -#![allow(dead_code)] - use crate::error::Error; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; use rustfs_filemeta::FileInfo; @@ -27,48 +24,11 @@ pub(crate) mod lifecycle { } pub(crate) mod list { - use super::{Debug, Error, FileInfo, ObjectInfo}; pub(crate) use rustfs_storage_api::{ ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsInfo, ListObjectsV2Info as StorageListObjectsV2Info, ListOperations, ObjectInfoOrErr as StorageObjectInfoOrErr, VersionMarker, WalkOptions as StorageWalkOptions, WalkVersionsSortOrder, }; - use tokio_util::sync::CancellationToken; - - type ListObjectsV2Info = StorageListObjectsV2Info; - type ListObjectVersionsInfo = StorageListObjectVersionsInfo; - type ObjectInfoOrErr = StorageObjectInfoOrErr; - type WalkOptions = StorageWalkOptions bool>; - - pub(crate) trait EcstoreListOperations: - ListOperations< - Error = Error, - ListObjectsV2Info = ListObjectsV2Info, - ListObjectVersionsInfo = ListObjectVersionsInfo, - ObjectInfoOrErr = ObjectInfoOrErr, - WalkOptions = WalkOptions, - WalkCancellation = CancellationToken, - WalkResultSender = tokio::sync::mpsc::Sender, - > + Send - + Sync - + Debug - { - } - - impl EcstoreListOperations for T where - T: ListOperations< - Error = Error, - ListObjectsV2Info = ListObjectsV2Info, - ListObjectVersionsInfo = ListObjectVersionsInfo, - ObjectInfoOrErr = ObjectInfoOrErr, - WalkOptions = WalkOptions, - WalkCancellation = CancellationToken, - WalkResultSender = tokio::sync::mpsc::Sender, - > + Send - + Sync - + Debug - { - } } pub(crate) mod multipart {