refactor(ecstore): relocate NamespaceLocking + finalize God-Object split (backlog#822) (#4291)

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 <Contract> 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<bool> = 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<NamespaceLockWrapper> {
-        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(

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<NamespaceLockWrapper> {
+        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<Option<DiskStore>> {
+    pub(in crate::set_disk) async fn get_disks_internal(&self) -> Vec<Option<DiskStore>> {
         let rl = self.disks.read().await;
 
         rl.clone()
@@ -94,7 +134,7 @@ impl SetDisks {
         disks
     }
 
-    pub(super) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
+    pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
 
@@ -104,7 +144,7 @@ impl SetDisks {
         disks
     }
 
-    pub(super) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
+    pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
         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<Option<DiskStore>> {
+    pub(in crate::set_disk) async fn _get_local_disks(&self) -> Vec<Option<DiskStore>> {
         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<Option<DiskStore>>, bool)> {
+    pub(in crate::set_disk) async fn get_online_disk_with_healing(
+        &self,
+        incl_healing: bool,
+    ) -> Result<(Vec<Option<DiskStore>>, 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<Option<DiskStore>>, Vec<DiskInfo>, usize)> {

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;

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<ObjectInfo>;
-    type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
-    type ObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, Error>;
-    type WalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
-
-    pub(crate) trait EcstoreListOperations:
-        ListOperations<
-            Error = Error,
-            ListObjectsV2Info = ListObjectsV2Info,
-            ListObjectVersionsInfo = ListObjectVersionsInfo,
-            ObjectInfoOrErr = ObjectInfoOrErr,
-            WalkOptions = WalkOptions,
-            WalkCancellation = CancellationToken,
-            WalkResultSender = tokio::sync::mpsc::Sender<ObjectInfoOrErr>,
-        > + Send
-        + Sync
-        + Debug
-    {
-    }
-
-    impl<T> EcstoreListOperations for T where
-        T: ListOperations<
-                Error = Error,
-                ListObjectsV2Info = ListObjectsV2Info,
-                ListObjectVersionsInfo = ListObjectVersionsInfo,
-                ObjectInfoOrErr = ObjectInfoOrErr,
-                WalkOptions = WalkOptions,
-                WalkCancellation = CancellationToken,
-                WalkResultSender = tokio::sync::mpsc::Sender<ObjectInfoOrErr>,
-            > + Send
-            + Sync
-            + Debug
-    {
-    }
 }
 
 pub(crate) mod multipart {