refactor: move heal and namespace contracts (#3560)

crates/ecstore/src/bucket/metadata_sys.rs

@@ -21,11 +21,11 @@ use crate::bucket::utils::{deserialize, is_meta_bucketname};
 use crate::error::{Error, Result, is_err_bucket_not_found};
 use crate::global::{GLOBAL_Endpoints, is_dist_erasure, is_erasure, resolve_object_store_handle};
 use crate::store::ECStore;
-use crate::store_api::HealOperations as _;
 use futures::future::join_all;
 use lazy_static::lazy_static;
 use rustfs_common::heal_channel::HealOpts;
 use rustfs_policy::policy::BucketPolicy;
+use rustfs_storage_api::HealOperations as _;
 use s3s::dto::ReplicationConfiguration;
 use s3s::dto::{
     AccelerateConfiguration, BucketLifecycleConfiguration, BucketLoggingStatus, CORSConfiguration, NotificationConfiguration,

crates/ecstore/src/config/com.rs

@@ -1217,7 +1217,7 @@ mod tests {
     use crate::error::{Error, Result};
     use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type};
     use crate::set_disk::SetDisks;
-    use crate::store_api::{GetObjectReader, NamespaceLocking, ObjectInfo, ObjectOptions, PutObjReader};
+    use crate::store_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
     use http::HeaderMap;
     use rustfs_config::audit::{AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS};
     use rustfs_config::notify::{
@@ -1231,6 +1231,7 @@ mod tests {
     use rustfs_lock::client::LockClient;
     use rustfs_lock::client::local::LocalClient;
     use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};
+    use rustfs_storage_api::NamespaceLocking as _;
     use rustfs_storage_api::{HTTPRangeSpec, StorageAdminApi};
     use serde_json::Value;
     use serial_test::serial;
@@ -1459,7 +1460,10 @@ mod tests {
     }
 
     #[async_trait::async_trait]
-    impl NamespaceLocking for LockingConfigStorage {
+    impl rustfs_storage_api::NamespaceLocking for LockingConfigStorage {
+        type Error = crate::error::Error;
+        type NamespaceLock = rustfs_lock::NamespaceLockWrapper;
+
         async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<rustfs_lock::NamespaceLockWrapper> {
             self.set_disks.new_ns_lock(bucket, object).await
         }

crates/ecstore/src/pools.rs

@@ -38,7 +38,7 @@ use crate::error::{
 use crate::notification_sys::get_global_notification_sys;
 use crate::resolve_object_store_handle;
 use crate::set_disk::SetDisks;
-use crate::store_api::{GetObjectReader, HealOperations, ObjectOptions};
+use crate::store_api::{GetObjectReader, ObjectOptions};
 use crate::{global::GLOBAL_LifecycleSys, sets::Sets, store::ECStore};
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
 use futures::{StreamExt, future::BoxFuture, stream::FuturesUnordered};
@@ -51,7 +51,8 @@ use rustfs_common::heal_channel::HealOpts;
 use rustfs_concurrency::workers::Workers;
 use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
 use rustfs_storage_api::{
-    BucketOperations, BucketOptions, MakeBucketOptions, ObjectIO as _, ObjectOperations as _, StorageAdminApi,
+    BucketOperations, BucketOptions, HealOperations as _, MakeBucketOptions, ObjectIO as _, ObjectOperations as _,
+    StorageAdminApi,
 };
 use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path};
 use s3s::dto::{BucketLifecycleConfiguration, DefaultRetention, ReplicationConfiguration};

crates/ecstore/src/set_disk.rs

@@ -52,8 +52,8 @@ use crate::{
     event_notification::{EventArgs, send_event},
     global::{GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_deployment_id, is_dist_erasure},
     store_api::{
-        DeletedObject, GetObjectReader, HealOperations, ListObjectsV2Info, MultipartOperations, NamespaceLocking, ObjectIO,
-        ObjectInfo, ObjectOperations, PutObjReader,
+        DeletedObject, GetObjectReader, ListObjectsV2Info, MultipartOperations, ObjectIO, ObjectInfo, ObjectOperations,
+        PutObjReader,
     },
     store_init::load_format_erasure,
 };
@@ -90,7 +90,7 @@ use rustfs_storage_api::{
     BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, ListMultipartsInfo, ListPartsInfo,
     MakeBucketOptions, MultipartInfo, MultipartUploadResult, PartInfo,
 };
-use rustfs_storage_api::{MultipartOperations as _, ObjectIO as _, ObjectOperations as _};
+use rustfs_storage_api::{MultipartOperations as _, NamespaceLocking as _, ObjectIO as _, ObjectOperations as _};
 use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
 use rustfs_utils::http::headers::AMZ_STORAGE_CLASS;
 use rustfs_utils::http::headers::{
@@ -1729,7 +1729,10 @@ impl SetDisks {
 }
 
 #[async_trait::async_trait]
-impl NamespaceLocking for SetDisks {
+impl rustfs_storage_api::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 is_dist_erasure().await {
@@ -4144,7 +4147,11 @@ impl rustfs_storage_api::MultipartOperations for SetDisks {
 }
 
 #[async_trait::async_trait]
-impl HealOperations for SetDisks {
+impl rustfs_storage_api::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>)> {
         unimplemented!()
@@ -5019,7 +5026,7 @@ mod tests {
     use rustfs_filemeta::ReplicationState;
     use rustfs_lock::client::local::LocalClient;
     use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};
-    use rustfs_storage_api::{CompletePart, ObjectOperations as _};
+    use rustfs_storage_api::{CompletePart, NamespaceLocking as _, ObjectOperations as _};
     use serial_test::serial;
     use std::collections::HashMap;
     use tempfile::TempDir;

crates/ecstore/src/set_disk/heal.rs

@@ -14,6 +14,7 @@
 
 use super::*;
 use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
+use rustfs_storage_api::NamespaceLocking as _;
 
 impl SetDisks {
     #[tracing::instrument(skip(self, opts), fields(bucket = %bucket, object = %object, version_id = %version_id))]

crates/ecstore/src/sets.rs

@@ -28,8 +28,8 @@ use crate::{
     global::{GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_lock_clients, is_dist_erasure},
     set_disk::SetDisks,
     store_api::{
-        DeletedObject, GetObjectReader, HealOperations, ListObjectVersionsInfo, ListObjectsV2Info, NamespaceLocking, ObjectInfo,
-        ObjectOptions, ObjectToDelete, PutObjReader,
+        DeletedObject, GetObjectReader, ListObjectVersionsInfo, ListObjectsV2Info, ObjectInfo, ObjectOptions, ObjectToDelete,
+        PutObjReader,
     },
     store_init::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
 };
@@ -826,7 +826,11 @@ impl rustfs_storage_api::MultipartOperations for Sets {
 }
 
 #[async_trait::async_trait]
-impl HealOperations for Sets {
+impl rustfs_storage_api::HealOperations for Sets {
+    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, _) = init_storage_disks_with_errors(
@@ -938,7 +942,10 @@ impl HealOperations for Sets {
 }
 
 #[async_trait::async_trait]
-impl NamespaceLocking for Sets {
+impl rustfs_storage_api::NamespaceLocking for Sets {
+    type Error = Error;
+    type NamespaceLock = NamespaceLockWrapper;
+
     async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper> {
         self.disk_set[0].new_ns_lock(bucket, object).await
     }

crates/ecstore/src/store.rs

@@ -62,10 +62,7 @@ use crate::{
     endpoints::EndpointServerPools,
     rpc::S3PeerSys,
     sets::Sets,
-    store_api::{
-        DeletedObject, GetObjectReader, HealOperations, ListObjectsV2Info, NamespaceLocking, ObjectInfo, ObjectOptions,
-        ObjectToDelete, PutObjReader,
-    },
+    store_api::{DeletedObject, GetObjectReader, ListObjectsV2Info, ObjectInfo, ObjectOptions, ObjectToDelete, PutObjReader},
     store_init,
 };
 use futures::future::join_all;
@@ -708,7 +705,11 @@ impl rustfs_storage_api::MultipartOperations for ECStore {
 }
 
 #[async_trait::async_trait]
-impl HealOperations for ECStore {
+impl rustfs_storage_api::HealOperations for ECStore {
+    type Error = Error;
+    type HealResultItem = HealResultItem;
+    type HealOptions = HealOpts;
+
     #[instrument(skip(self))]
     async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
         self.handle_heal_format(dry_run).await
@@ -741,7 +742,10 @@ impl HealOperations for ECStore {
 }
 
 #[async_trait::async_trait]
-impl NamespaceLocking for ECStore {
+impl rustfs_storage_api::NamespaceLocking for ECStore {
+    type Error = Error;
+    type NamespaceLock = NamespaceLockWrapper;
+
     async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper> {
         self.handle_new_ns_lock(bucket, object).await
     }

crates/ecstore/src/store/bucket.rs

@@ -19,6 +19,7 @@ use crate::bucket::{
 };
 use crate::global::get_global_bucket_monitor;
 use crate::set_disk::get_lock_acquire_timeout;
+use rustfs_storage_api::NamespaceLocking as _;
 
 fn should_override_created_from_metadata(created: OffsetDateTime) -> bool {
     created != OffsetDateTime::UNIX_EPOCH

crates/ecstore/src/store/heal.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use super::*;
+use rustfs_storage_api::HealOperations as _;
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_HEAL: &str = "heal";

crates/ecstore/src/store/rebalance.rs

@@ -14,7 +14,7 @@
 
 use super::*;
 use crate::config::get_global_storage_class;
-use rustfs_storage_api::{ObjectOperations as _, StorageAdminApi};
+use rustfs_storage_api::{NamespaceLocking as _, ObjectOperations as _, StorageAdminApi};
 
 struct LatestObjectInfoCandidate {
     info: Option<ObjectInfo>,

crates/ecstore/src/store_api/traits.rs

@@ -131,24 +131,27 @@ impl<T> MultipartOperations for T where
 }
 
 /// Healing and repair operations.
-#[async_trait::async_trait]
-pub trait HealOperations: Send + Sync + Debug {
-    async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)>;
-    async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem>;
-    async fn heal_object(
-        &self,
-        bucket: &str,
-        object: &str,
-        version_id: &str,
-        opts: &HealOpts,
-    ) -> Result<(HealResultItem, Option<Error>)>;
-    async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>)>;
-    async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()>;
+pub trait HealOperations:
+    rustfs_storage_api::HealOperations<Error = Error, HealResultItem = HealResultItem, HealOptions = HealOpts> + Send + Sync + Debug
+{
+}
+
+impl<T> HealOperations for T where
+    T: rustfs_storage_api::HealOperations<Error = Error, HealResultItem = HealResultItem, HealOptions = HealOpts>
+        + Send
+        + Sync
+        + Debug
+{
 }
 
 /// Namespace lock operations needed by consumers that only coordinate object
 /// mutations but do not require the full storage API surface.
-#[async_trait::async_trait]
-pub trait NamespaceLocking: Send + Sync + Debug + 'static {
-    async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<NamespaceLockWrapper>;
+pub trait NamespaceLocking:
+    rustfs_storage_api::NamespaceLocking<Error = Error, NamespaceLock = NamespaceLockWrapper> + Send + Sync + Debug + 'static
+{
+}
+
+impl<T> NamespaceLocking for T where
+    T: rustfs_storage_api::NamespaceLocking<Error = Error, NamespaceLock = NamespaceLockWrapper> + Send + Sync + Debug + 'static
+{
 }

crates/ecstore/tests/ecstore_contract_compat_test.rs

@@ -12,8 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use rustfs_ecstore::{disk::DiskStore, error::Error, store::ECStore, store_api::NamespaceLocking};
-use rustfs_storage_api::StorageAdminApi;
+use rustfs_common::heal_channel::HealOpts;
+use rustfs_ecstore::{
+    disk::DiskStore,
+    error::Error,
+    store::ECStore,
+    store_api::{HealOperations, NamespaceLocking},
+};
+use rustfs_lock::NamespaceLockWrapper;
+use rustfs_madmin::heal_commands::HealResultItem;
+use rustfs_storage_api::{HealOperations as StorageHealOperations, NamespaceLocking as StorageNamespaceLocking, StorageAdminApi};
 
 fn storage_admin_api_type_name<T>() -> &'static str
 where
@@ -34,6 +42,27 @@ where
     std::any::type_name::<T>()
 }
 
+fn heal_operations_type_name<T>() -> &'static str
+where
+    T: HealOperations,
+{
+    std::any::type_name::<T>()
+}
+
+fn storage_namespace_locking_type_name<T>() -> &'static str
+where
+    T: StorageNamespaceLocking<Error = Error, NamespaceLock = NamespaceLockWrapper>,
+{
+    std::any::type_name::<T>()
+}
+
+fn storage_heal_operations_type_name<T>() -> &'static str
+where
+    T: StorageHealOperations<Error = Error, HealResultItem = HealResultItem, HealOptions = HealOpts>,
+{
+    std::any::type_name::<T>()
+}
+
 #[test]
 fn ecstore_implements_storage_admin_api_contract() {
     assert!(storage_admin_api_type_name::<ECStore>().ends_with("::ECStore"));
@@ -43,3 +72,18 @@ fn ecstore_implements_storage_admin_api_contract() {
 fn ecstore_implements_namespace_locking_contract() {
     assert!(namespace_locking_type_name::<ECStore>().ends_with("::ECStore"));
 }
+
+#[test]
+fn ecstore_implements_heal_operations_contract() {
+    assert!(heal_operations_type_name::<ECStore>().ends_with("::ECStore"));
+}
+
+#[test]
+fn ecstore_implements_storage_namespace_locking_contract() {
+    assert!(storage_namespace_locking_type_name::<ECStore>().ends_with("::ECStore"));
+}
+
+#[test]
+fn ecstore_implements_storage_heal_operations_contract() {
+    assert!(storage_heal_operations_type_name::<ECStore>().ends_with("::ECStore"));
+}

crates/heal/src/heal/storage.rs

@@ -19,11 +19,12 @@ use rustfs_ecstore::{
     disk::{DiskStore, endpoint::Endpoint},
     error::StorageError,
     store::ECStore,
-    store_api::{HealOperations, ObjectOptions},
+    store_api::ObjectOptions,
 };
 use rustfs_madmin::heal_commands::HealResultItem;
 use rustfs_storage_api::{
-    BucketInfo, BucketOperations, DiskSetSelector, ListOperations as _, ObjectIO as _, ObjectOperations as _, StorageAdminApi,
+    BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _,
+    ObjectOperations as _, StorageAdminApi,
 };
 use std::sync::Arc;
 use tracing::{debug, error, warn};

crates/scanner/src/scanner.rs

@@ -45,8 +45,8 @@ use rustfs_ecstore::disk::RUSTFS_META_BUCKET;
 use rustfs_ecstore::error::Error as EcstoreError;
 use rustfs_ecstore::global::is_erasure_sd;
 use rustfs_ecstore::store::ECStore;
-use rustfs_ecstore::store_api::{NamespaceLocking as _, ObjectIO};
-use rustfs_storage_api::{BucketOperations, BucketOptions};
+use rustfs_ecstore::store_api::ObjectIO;
+use rustfs_storage_api::{BucketOperations, BucketOptions, NamespaceLocking as _};
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 use tokio::time::{Duration, Instant};

crates/storage-api/src/lib.rs

@@ -25,7 +25,7 @@ pub use bucket::{BucketInfo, BucketOperations, BucketOptions, DeleteBucketOption
 pub use error::{StorageErrorCode, StorageResult};
 pub use multipart::{CompletePart, ListMultipartsInfo, ListPartsInfo, MultipartInfo, MultipartUploadResult, PartInfo};
 pub use object::{HTTPPreconditions, HTTPRangeError, HTTPRangeSpec, ObjectLockRetentionOptions};
+pub use object::{HealOperations, MultipartOperations, NamespaceLocking, ObjectIO, ObjectOperations};
 pub use object::{ListObjectVersionsInfo, ListObjectsInfo, ListObjectsV2Info, ListOperations, ObjectInfoOrErr};
-pub use object::{MultipartOperations, ObjectIO, ObjectOperations};
 pub use object::{ObjectPreconditionError, ObjectPreconditionPart, ObjectPreconditionState};
 pub use object::{VersionMarker, WalkOptions, WalkVersionsSortOrder};

crates/storage-api/src/object.rs

@@ -408,6 +408,33 @@ pub trait MultipartOperations: Send + Sync + fmt::Debug {
     ) -> Result<Self::ObjectInfo, Self::Error>;
 }
 
+#[async_trait::async_trait]
+pub trait HealOperations: Send + Sync + fmt::Debug {
+    type Error: std::error::Error + Send + Sync + 'static;
+    type HealResultItem: Send + 'static;
+    type HealOptions: Send + Sync + 'static;
+
+    async fn heal_format(&self, dry_run: bool) -> Result<(Self::HealResultItem, Option<Self::Error>), Self::Error>;
+    async fn heal_bucket(&self, bucket: &str, opts: &Self::HealOptions) -> Result<Self::HealResultItem, Self::Error>;
+    async fn heal_object(
+        &self,
+        bucket: &str,
+        object: &str,
+        version_id: &str,
+        opts: &Self::HealOptions,
+    ) -> Result<(Self::HealResultItem, Option<Self::Error>), Self::Error>;
+    async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>), Self::Error>;
+    async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &Self::HealOptions) -> Result<(), Self::Error>;
+}
+
+#[async_trait::async_trait]
+pub trait NamespaceLocking: Send + Sync + fmt::Debug + 'static {
+    type Error: std::error::Error + Send + Sync + 'static;
+    type NamespaceLock: Send + 'static;
+
+    async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result<Self::NamespaceLock, Self::Error>;
+}
+
 #[derive(Debug, Default)]
 pub struct ListObjectsInfo<ObjectItem> {
     pub is_truncated: bool,