mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
refactor: clean scanner heal runtime boundaries (#3571)
This commit is contained in:
@@ -14,6 +14,8 @@
|
||||
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::heal::storage_compat::{DiskError, EcstoreError};
|
||||
|
||||
/// Custom error type for heal operations
|
||||
/// This enum defines various error variants that can occur during
|
||||
/// the execution of heal-related tasks, such as I/O errors, storage errors,
|
||||
@@ -24,10 +26,10 @@ pub enum Error {
|
||||
Io(#[from] std::io::Error),
|
||||
|
||||
#[error("Storage error: {0}")]
|
||||
Storage(#[from] rustfs_ecstore::error::Error),
|
||||
Storage(#[from] EcstoreError),
|
||||
|
||||
#[error("Disk error: {0}")]
|
||||
Disk(#[from] rustfs_ecstore::disk::error::DiskError),
|
||||
Disk(#[from] DiskError),
|
||||
|
||||
#[error("Configuration error: {0}")]
|
||||
Config(String),
|
||||
|
||||
@@ -489,6 +489,7 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::heal::manager::HealConfig;
|
||||
use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
|
||||
use crate::heal::storage_compat::{DiskStore, Endpoint};
|
||||
use rustfs_common::heal_channel::{
|
||||
HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealRequestSource, HealScanMode,
|
||||
};
|
||||
@@ -516,13 +517,10 @@ mod tests {
|
||||
async fn ec_decode_rebuild(&self, _bucket: &str, _object: &str) -> crate::Result<Vec<u8>> {
|
||||
Ok(vec![])
|
||||
}
|
||||
async fn get_disk_status(
|
||||
&self,
|
||||
_endpoint: &rustfs_ecstore::disk::endpoint::Endpoint,
|
||||
) -> crate::Result<crate::heal::storage::DiskStatus> {
|
||||
async fn get_disk_status(&self, _endpoint: &Endpoint) -> crate::Result<crate::heal::storage::DiskStatus> {
|
||||
Ok(crate::heal::storage::DiskStatus::Ok)
|
||||
}
|
||||
async fn format_disk(&self, _endpoint: &rustfs_ecstore::disk::endpoint::Endpoint) -> crate::Result<()> {
|
||||
async fn format_disk(&self, _endpoint: &Endpoint) -> crate::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
async fn get_bucket_info(&self, _bucket: &str) -> crate::Result<Option<rustfs_storage_api::BucketInfo>> {
|
||||
@@ -576,7 +574,7 @@ mod tests {
|
||||
) -> crate::Result<(Vec<String>, Option<String>, bool)> {
|
||||
Ok((vec![], None, false))
|
||||
}
|
||||
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> crate::Result<rustfs_ecstore::disk::DiskStore> {
|
||||
async fn get_disk_for_resume(&self, _set_disk_id: &str) -> crate::Result<DiskStore> {
|
||||
Err(crate::Error::other("Not implemented in mock"))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,7 +21,6 @@ use crate::{Error, Result};
|
||||
use futures::{StreamExt, future::join_all, stream::FuturesUnordered};
|
||||
use metrics::gauge;
|
||||
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
|
||||
use rustfs_ecstore::disk::DiskStore;
|
||||
use std::sync::{
|
||||
Arc,
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
@@ -29,6 +28,8 @@ use std::sync::{
|
||||
use tokio::sync::{RwLock, Semaphore};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use super::storage_compat::DiskStore;
|
||||
|
||||
const LOG_COMPONENT_HEAL: &str = "heal";
|
||||
const LOG_SUBSYSTEM_ERASURE_HEALER: &str = "erasure_healer";
|
||||
const EVENT_HEAL_ERASURE_RESUME_STATE: &str = "heal_erasure_resume_state";
|
||||
|
||||
@@ -14,10 +14,11 @@
|
||||
|
||||
use crate::heal::{HealOptions, HealPriority, HealRequest, HealType};
|
||||
use crate::{Error, Result};
|
||||
use rustfs_ecstore::disk::endpoint::Endpoint;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::time::SystemTime;
|
||||
|
||||
use super::storage_compat::Endpoint;
|
||||
|
||||
/// Corruption type
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum CorruptionType {
|
||||
|
||||
@@ -20,9 +20,6 @@ use crate::heal::{
|
||||
use crate::{Error, Result};
|
||||
use metrics::{counter, gauge};
|
||||
use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource};
|
||||
use rustfs_ecstore::disk::DiskAPI;
|
||||
use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
|
||||
use rustfs_madmin::heal_commands::HealResultItem;
|
||||
use std::{
|
||||
collections::{BinaryHeap, HashMap},
|
||||
@@ -36,6 +33,8 @@ use tokio::{
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use super::storage_compat::{DiskAPI, DiskError, GLOBAL_LOCAL_DISK_MAP};
|
||||
|
||||
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
|
||||
const LOG_COMPONENT_HEAL: &str = "heal";
|
||||
const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
|
||||
@@ -2286,10 +2285,11 @@ mod tests {
|
||||
use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
|
||||
use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
|
||||
use rustfs_common::heal_channel::{HealOpts, HealRequestSource};
|
||||
use rustfs_ecstore::disk::{DiskStore, endpoint::Endpoint};
|
||||
use rustfs_madmin::heal_commands::HealResultItem;
|
||||
use rustfs_storage_api::BucketInfo;
|
||||
|
||||
use super::super::storage_compat::{DiskStore, Endpoint};
|
||||
|
||||
struct MockStorage;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
|
||||
@@ -19,7 +19,7 @@ pub mod manager;
|
||||
pub mod progress;
|
||||
pub mod resume;
|
||||
pub mod storage;
|
||||
mod storage_compat;
|
||||
pub(crate) mod storage_compat;
|
||||
pub mod task;
|
||||
pub mod utils;
|
||||
|
||||
|
||||
@@ -13,8 +13,6 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::{Error, Result};
|
||||
use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::disk::{BUCKET_META_PREFIX, DiskAPI, DiskStore, RUSTFS_META_BUCKET};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
@@ -23,6 +21,8 @@ use tokio::sync::RwLock;
|
||||
use tracing::{debug, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::storage_compat::{BUCKET_META_PREFIX, DiskAPI, DiskError, DiskStore, RUSTFS_META_BUCKET};
|
||||
|
||||
const LOG_COMPONENT_HEAL: &str = "heal";
|
||||
const LOG_SUBSYSTEM_RESUME: &str = "resume";
|
||||
const EVENT_HEAL_RESUME_STATE: &str = "heal_resume_state";
|
||||
@@ -744,7 +744,7 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_resumable_tasks_integration() {
|
||||
use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk};
|
||||
use super::super::storage_compat::{DiskOption, Endpoint, new_disk};
|
||||
use tempfile::TempDir;
|
||||
|
||||
// Create a temporary directory for testing
|
||||
|
||||
@@ -15,11 +15,6 @@
|
||||
use crate::{Error, Result};
|
||||
use async_trait::async_trait;
|
||||
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
|
||||
use rustfs_ecstore::{
|
||||
disk::{DiskStore, endpoint::Endpoint},
|
||||
error::StorageError,
|
||||
store::ECStore,
|
||||
};
|
||||
use rustfs_madmin::heal_commands::HealResultItem;
|
||||
use rustfs_storage_api::{
|
||||
BucketInfo, BucketOperations, DiskSetSelector, HealOperations as _, ListOperations as _, ObjectIO as _,
|
||||
@@ -28,6 +23,7 @@ use rustfs_storage_api::{
|
||||
use std::sync::Arc;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use super::storage_compat::{DiskStore, ECStore, Endpoint, StorageError};
|
||||
pub use super::storage_compat::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
|
||||
|
||||
const LOG_COMPONENT_HEAL: &str = "heal";
|
||||
@@ -200,7 +196,7 @@ impl HealStorageAPI for ECStoreHealStorage {
|
||||
Ok(info) => Ok(Some(info)),
|
||||
Err(e) => {
|
||||
// Map ObjectNotFound to None to align with Option return type
|
||||
if matches!(e, rustfs_ecstore::error::StorageError::ObjectNotFound(_, _)) {
|
||||
if matches!(e, StorageError::ObjectNotFound(_, _)) {
|
||||
debug!(
|
||||
target: "rustfs::heal::storage",
|
||||
event = EVENT_HEAL_STORAGE_OBJECT_IO,
|
||||
@@ -810,7 +806,7 @@ impl HealStorageAPI for ECStoreHealStorage {
|
||||
match self.ecstore.get_object_info(bucket, object, &opts).await {
|
||||
Ok(_) => Ok(true), // Object exists
|
||||
Err(e) => {
|
||||
if matches!(e, rustfs_ecstore::error::StorageError::ObjectNotFound(_, _)) {
|
||||
if matches!(e, StorageError::ObjectNotFound(_, _)) {
|
||||
debug!(
|
||||
target: "rustfs::heal::storage",
|
||||
event = EVENT_HEAL_STORAGE_OBJECT_IO,
|
||||
@@ -1236,8 +1232,8 @@ impl HealStorageAPI for ECStoreHealStorage {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::storage_compat::StorageError;
|
||||
use super::{is_transient_object_exists_error, is_transient_object_exists_message};
|
||||
use rustfs_ecstore::error::StorageError;
|
||||
|
||||
#[test]
|
||||
fn transient_object_exists_message_matches_lock_quorum_failures() {
|
||||
|
||||
@@ -12,10 +12,18 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use rustfs_ecstore::store_api::{
|
||||
ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions, PutObjReader as EcstorePutObjReader,
|
||||
pub(crate) use rustfs_ecstore::{
|
||||
data_usage::DATA_USAGE_CACHE_NAME,
|
||||
disk::{BUCKET_META_PREFIX, DiskAPI, DiskStore, RUSTFS_META_BUCKET, endpoint::Endpoint, error::DiskError},
|
||||
error::{Error as EcstoreError, StorageError},
|
||||
global::GLOBAL_LOCAL_DISK_MAP,
|
||||
store::ECStore,
|
||||
store_api::{ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions, PutObjReader as EcstorePutObjReader},
|
||||
};
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::disk::{DiskOption, new_disk};
|
||||
|
||||
pub type HealObjectInfo = EcstoreObjectInfo;
|
||||
pub type HealObjectOptions = EcstoreObjectOptions;
|
||||
pub type HealPutObjReader = EcstorePutObjReader;
|
||||
|
||||
@@ -16,10 +16,6 @@ use crate::heal::{ErasureSetHealer, progress::HealProgress, storage::HealStorage
|
||||
use crate::{Error, Result};
|
||||
use metrics::{counter, histogram};
|
||||
use rustfs_common::heal_channel::{HealOpts, HealRequestSource, HealScanMode};
|
||||
use rustfs_ecstore::{
|
||||
data_usage::DATA_USAGE_CACHE_NAME,
|
||||
disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
|
||||
};
|
||||
use rustfs_madmin::heal_commands::HealResultItem;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
@@ -31,6 +27,8 @@ use tokio::sync::RwLock;
|
||||
use tracing::{debug, error, info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use super::storage_compat::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
|
||||
|
||||
const LOG_COMPONENT_HEAL: &str = "heal";
|
||||
const LOG_SUBSYSTEM_TASK: &str = "task";
|
||||
const LOG_SUBSYSTEM_OBJECT: &str = "object";
|
||||
@@ -2049,12 +2047,9 @@ impl std::fmt::Debug for HealTask {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::storage_compat::{DiskStore, Endpoint};
|
||||
use super::*;
|
||||
use crate::heal::storage::{DiskStatus, HealObjectInfo};
|
||||
use rustfs_ecstore::{
|
||||
data_usage::DATA_USAGE_CACHE_NAME,
|
||||
disk::{BUCKET_META_PREFIX, DiskStore, RUSTFS_META_BUCKET, endpoint::Endpoint},
|
||||
};
|
||||
use rustfs_madmin::heal_commands::HealResultItem;
|
||||
use rustfs_storage_api::BucketInfo;
|
||||
use std::sync::Mutex;
|
||||
|
||||
@@ -28,20 +28,18 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
|
||||
pub use rustfs_data_usage::{
|
||||
BucketTargetUsageInfo, BucketUsageInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, hash_path,
|
||||
};
|
||||
use rustfs_ecstore::{
|
||||
bucket::{lifecycle::lifecycle::TRANSITION_COMPLETE, replication::ReplicationConfig},
|
||||
config::{com::save_config, storageclass},
|
||||
disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
|
||||
error::{Error, Result as StorageResult, StorageError},
|
||||
};
|
||||
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
|
||||
use tokio::time::{Duration, Instant, sleep, timeout};
|
||||
use tracing::warn;
|
||||
|
||||
use crate::storage_compat::{
|
||||
BUCKET_META_PREFIX, EcstoreError as Error, EcstoreResult as StorageResult, RUSTFS_META_BUCKET, ReplicationConfig,
|
||||
ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, StorageError, TRANSITION_COMPLETE, save_config,
|
||||
storageclass,
|
||||
};
|
||||
pub use crate::storage_compat::{
|
||||
ScannerGetObjectReader, ScannerObjectIO, ScannerObjectInfo, ScannerObjectOptions, ScannerObjectToDelete, ScannerPutObjReader,
|
||||
};
|
||||
use crate::storage_compat::{ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions};
|
||||
|
||||
// Data usage constants
|
||||
pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
|
||||
|
||||
@@ -817,6 +817,7 @@ pub(crate) fn scanner_alert_excess_folders() -> u64 {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{ScannerRuntimeConfigSource, lookup_scanner_runtime_config, validate_scanner_runtime_config};
|
||||
use crate::storage_compat::init_ecstore_config_for_scanner_tests;
|
||||
use rustfs_config::server_config::{Config as ServerConfig, KVS};
|
||||
use rustfs_config::{
|
||||
DEFAULT_DELIMITER, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE,
|
||||
@@ -831,7 +832,7 @@ mod tests {
|
||||
use temp_env::{with_var, with_var_unset};
|
||||
|
||||
fn server_config_with_scanner(entries: &[(&str, &str)]) -> ServerConfig {
|
||||
rustfs_ecstore::config::init();
|
||||
init_ecstore_config_for_scanner_tests();
|
||||
let mut config = ServerConfig::new();
|
||||
let mut kvs = KVS::new();
|
||||
for (key, value) in entries {
|
||||
|
||||
@@ -39,14 +39,6 @@ use rustfs_config::{
|
||||
ENV_SCANNER_CYCLE_MAX_OBJECTS,
|
||||
};
|
||||
use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
|
||||
use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _;
|
||||
use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_replication_config};
|
||||
use rustfs_ecstore::bucket::replication::ReplicationConfigurationExt as _;
|
||||
use rustfs_ecstore::config::com::{read_config, save_config};
|
||||
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_storage_api::{BucketOperations, BucketOptions, NamespaceLocking as _};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc;
|
||||
@@ -54,6 +46,11 @@ use tokio::time::{Duration, Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, instrument, warn};
|
||||
|
||||
use crate::storage_compat::{
|
||||
ECStore, EcstoreError, Lifecycle as _, RUSTFS_META_BUCKET, ReplicationConfigurationExt as _, get_lifecycle_config,
|
||||
get_replication_config, is_erasure_sd, read_config, replace_bucket_usage_memory_from_info, save_config,
|
||||
};
|
||||
|
||||
const LOG_COMPONENT_SCANNER: &str = "scanner";
|
||||
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
|
||||
const LOG_SUBSYSTEM_BACKGROUND_HEAL: &str = "background_heal";
|
||||
@@ -1059,7 +1056,7 @@ pub async fn store_data_usage_in_backend(
|
||||
"Scanner data usage save failed"
|
||||
);
|
||||
} else {
|
||||
rustfs_ecstore::data_usage::replace_bucket_usage_memory_from_info(&data_usage_info).await;
|
||||
replace_bucket_usage_memory_from_info(&data_usage_info).await;
|
||||
}
|
||||
done_save();
|
||||
|
||||
@@ -1070,6 +1067,7 @@ pub async fn store_data_usage_in_backend(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::storage_compat::EcstoreResult;
|
||||
use crate::{
|
||||
ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
|
||||
ScannerPutObjReader as PutObjReader,
|
||||
@@ -1124,7 +1122,7 @@ mod tests {
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl rustfs_storage_api::ObjectIO for MemoryConfigStore {
|
||||
type Error = rustfs_ecstore::error::Error;
|
||||
type Error = EcstoreError;
|
||||
type RangeSpec = rustfs_storage_api::HTTPRangeSpec;
|
||||
type HeaderMap = http::HeaderMap;
|
||||
type ObjectOptions = ObjectOptions;
|
||||
@@ -1139,12 +1137,12 @@ mod tests {
|
||||
_range: Option<rustfs_storage_api::HTTPRangeSpec>,
|
||||
_h: http::HeaderMap,
|
||||
_opts: &ObjectOptions,
|
||||
) -> rustfs_ecstore::error::Result<GetObjectReader> {
|
||||
) -> EcstoreResult<GetObjectReader> {
|
||||
let objects = self.objects.lock().await;
|
||||
let data = objects
|
||||
.get(&memory_config_key(bucket, object))
|
||||
.cloned()
|
||||
.ok_or(rustfs_ecstore::error::Error::FileNotFound)?;
|
||||
.ok_or(EcstoreError::FileNotFound)?;
|
||||
|
||||
Ok(GetObjectReader {
|
||||
stream: Box::new(Cursor::new(data)),
|
||||
@@ -1158,7 +1156,7 @@ mod tests {
|
||||
object: &str,
|
||||
data: &mut PutObjReader,
|
||||
_opts: &ObjectOptions,
|
||||
) -> rustfs_ecstore::error::Result<ObjectInfo> {
|
||||
) -> EcstoreResult<ObjectInfo> {
|
||||
let mut buf = Vec::new();
|
||||
data.stream.read_to_end(&mut buf).await?;
|
||||
self.objects.lock().await.insert(memory_config_key(bucket, object), buf);
|
||||
|
||||
@@ -39,25 +39,6 @@ use rustfs_common::metrics::{
|
||||
IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn,
|
||||
current_path_updater, global_metrics,
|
||||
};
|
||||
use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
|
||||
use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule};
|
||||
use rustfs_ecstore::bucket::lifecycle::evaluator::Evaluator;
|
||||
use rustfs_ecstore::bucket::lifecycle::{
|
||||
bucket_lifecycle_ops::apply_transition_rule,
|
||||
lifecycle::{Event, Lifecycle, ObjectOpts},
|
||||
};
|
||||
use rustfs_ecstore::bucket::replication::{
|
||||
ReplicationConfig, ReplicationConfigurationExt as _, ReplicationQueueAdmission, queue_replication_heal_internal,
|
||||
};
|
||||
use rustfs_ecstore::bucket::versioning::VersioningApi;
|
||||
use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
|
||||
use rustfs_ecstore::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
|
||||
use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::disk::{Disk, DiskAPI as _, DiskInfoOptions};
|
||||
use rustfs_ecstore::error::StorageError;
|
||||
use rustfs_ecstore::global::is_erasure;
|
||||
use rustfs_ecstore::pools::{path2_bucket_object, path2_bucket_object_with_base_path};
|
||||
use rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket;
|
||||
use rustfs_filemeta::{
|
||||
MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType,
|
||||
};
|
||||
@@ -69,6 +50,12 @@ use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::storage_compat::{
|
||||
BucketVersioningSys, Disk, DiskAPI as _, DiskError, DiskInfoOptions, Evaluator, Event, GLOBAL_ExpiryState, LcEventSrc,
|
||||
Lifecycle, ListPathRawOptions, ObjectOpts, ReplicationConfig, ReplicationConfigurationExt as _, ReplicationQueueAdmission,
|
||||
StorageError, VersioningApi, apply_expiry_rule, apply_transition_rule, is_erasure, is_reserved_or_invalid_bucket,
|
||||
list_path_raw, path2_bucket_object, path2_bucket_object_with_base_path, queue_replication_heal_internal,
|
||||
};
|
||||
use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
|
||||
|
||||
const LOG_COMPONENT_SCANNER: &str = "scanner";
|
||||
@@ -2436,7 +2423,7 @@ mod tests {
|
||||
use crate::SCANNER_SLEEPER;
|
||||
|
||||
use super::*;
|
||||
use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk};
|
||||
use crate::storage_compat::{DiskOption, Endpoint, new_disk};
|
||||
use rustfs_filemeta::{ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType};
|
||||
use serial_test::serial;
|
||||
#[cfg(unix)]
|
||||
|
||||
@@ -26,22 +26,6 @@ use rustfs_common::heal_channel::HealScanMode;
|
||||
use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics};
|
||||
#[cfg(test)]
|
||||
use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
|
||||
use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
|
||||
use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState;
|
||||
use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle;
|
||||
use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_object_lock_config, get_replication_config};
|
||||
use rustfs_ecstore::bucket::replication::{ReplicationConfig, ReplicationConfigurationExt};
|
||||
use rustfs_ecstore::bucket::versioning::VersioningApi as _;
|
||||
use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
|
||||
use rustfs_ecstore::config::storageclass;
|
||||
use rustfs_ecstore::disk::STORAGE_FORMAT_FILE;
|
||||
use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::disk::{Disk, DiskAPI};
|
||||
use rustfs_ecstore::error::{Error, StorageError};
|
||||
use rustfs_ecstore::global::GLOBAL_TierConfigMgr;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::set_disk::SetDisks;
|
||||
use rustfs_ecstore::{error::Result, store::ECStore};
|
||||
use rustfs_filemeta::FileMeta;
|
||||
use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
|
||||
use rustfs_utils::path::path_join_buf;
|
||||
@@ -59,6 +43,12 @@ use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::ScannerObjectInfo as ObjectInfo;
|
||||
use crate::storage_compat::{
|
||||
BucketTargetSys, BucketVersioningSys, Disk, DiskAPI, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
|
||||
GLOBAL_ExpiryState, GLOBAL_TierConfigMgr, Lifecycle, ReplicationConfig, ReplicationConfigurationExt, STORAGE_FORMAT_FILE,
|
||||
SetDisks, StorageError, VersioningApi as _, get_lifecycle_config, get_object_lock_config, get_replication_config,
|
||||
resolve_scanner_object_store_handle, storageclass,
|
||||
};
|
||||
|
||||
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
|
||||
const LOG_COMPONENT_SCANNER: &str = "scanner";
|
||||
@@ -1433,7 +1423,7 @@ impl ScannerIODisk for Disk {
|
||||
|
||||
// TODO: object lock
|
||||
|
||||
let Some(ecstore) = resolve_object_store_handle() else {
|
||||
let Some(ecstore) = resolve_scanner_object_store_handle() else {
|
||||
error!(
|
||||
target: "rustfs::scanner::io",
|
||||
event = EVENT_SCANNER_DISK_BUCKET_STATE,
|
||||
@@ -1529,8 +1519,7 @@ impl ScannerIODisk for Disk {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::scanner_folder::ScannerItem;
|
||||
use rustfs_ecstore::disk::{DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, new_disk};
|
||||
use rustfs_ecstore::pools::path2_bucket_object_with_base_path;
|
||||
use crate::storage_compat::{DiskOption, Endpoint, new_disk, path2_bucket_object_with_base_path};
|
||||
use serial_test::serial;
|
||||
use temp_env::with_var;
|
||||
use uuid::Uuid;
|
||||
|
||||
@@ -13,14 +13,48 @@
|
||||
// limitations under the License.
|
||||
|
||||
use http::HeaderMap;
|
||||
use rustfs_ecstore::{
|
||||
error::Error,
|
||||
pub(crate) use rustfs_ecstore::{
|
||||
bucket::{
|
||||
bucket_target_sys::BucketTargetSys,
|
||||
lifecycle::{
|
||||
bucket_lifecycle_audit::LcEventSrc,
|
||||
bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule, apply_transition_rule},
|
||||
evaluator::Evaluator,
|
||||
lifecycle::{Event, Lifecycle, ObjectOpts, TRANSITION_COMPLETE},
|
||||
},
|
||||
metadata_sys::{get_lifecycle_config, get_object_lock_config, get_replication_config},
|
||||
replication::{
|
||||
ReplicationConfig, ReplicationConfigurationExt, ReplicationQueueAdmission, queue_replication_heal_internal,
|
||||
},
|
||||
versioning::VersioningApi,
|
||||
versioning_sys::BucketVersioningSys,
|
||||
},
|
||||
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
|
||||
config::{
|
||||
com::{read_config, save_config},
|
||||
storageclass,
|
||||
},
|
||||
data_usage::replace_bucket_usage_memory_from_info,
|
||||
disk::{BUCKET_META_PREFIX, Disk, DiskAPI, DiskInfoOptions, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE, error::DiskError},
|
||||
error::{Error as EcstoreError, Result as EcstoreResult, StorageError},
|
||||
global::{GLOBAL_TierConfigMgr, is_erasure, is_erasure_sd},
|
||||
pools::{path2_bucket_object, path2_bucket_object_with_base_path},
|
||||
set_disk::SetDisks,
|
||||
store::ECStore,
|
||||
store_api::{
|
||||
GetObjectReader as EcstoreGetObjectReader, ObjectInfo as EcstoreObjectInfo, ObjectOptions as EcstoreObjectOptions,
|
||||
ObjectToDelete as EcstoreObjectToDelete, PutObjReader as EcstorePutObjReader,
|
||||
},
|
||||
store_utils::is_reserved_or_invalid_bucket,
|
||||
};
|
||||
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO};
|
||||
use std::sync::Arc;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::{
|
||||
config::init as init_ecstore_config_for_scanner_tests,
|
||||
disk::{DiskOption, endpoint::Endpoint, new_disk},
|
||||
};
|
||||
|
||||
pub type ScannerGetObjectReader = EcstoreGetObjectReader;
|
||||
pub type ScannerObjectInfo = EcstoreObjectInfo;
|
||||
@@ -28,9 +62,13 @@ pub type ScannerObjectOptions = EcstoreObjectOptions;
|
||||
pub type ScannerObjectToDelete = EcstoreObjectToDelete;
|
||||
pub type ScannerPutObjReader = EcstorePutObjReader;
|
||||
|
||||
pub(crate) fn resolve_scanner_object_store_handle() -> Option<Arc<ECStore>> {
|
||||
rustfs_ecstore::resolve_object_store_handle()
|
||||
}
|
||||
|
||||
pub trait ScannerObjectIO:
|
||||
ObjectIO<
|
||||
Error = Error,
|
||||
Error = EcstoreError,
|
||||
RangeSpec = HTTPRangeSpec,
|
||||
HeaderMap = HeaderMap,
|
||||
ObjectOptions = ScannerObjectOptions,
|
||||
@@ -43,7 +81,7 @@ pub trait ScannerObjectIO:
|
||||
|
||||
impl<T> ScannerObjectIO for T where
|
||||
T: ObjectIO<
|
||||
Error = Error,
|
||||
Error = EcstoreError,
|
||||
RangeSpec = HTTPRangeSpec,
|
||||
HeaderMap = HeaderMap,
|
||||
ObjectOptions = ScannerObjectOptions,
|
||||
|
||||
@@ -844,6 +844,26 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
|
||||
formatting, diff hygiene, direct Swift import scan, Rust risk scan, full
|
||||
pre-commit, and required three-expert review passed.
|
||||
|
||||
- [x] `API-029` Clean scanner and heal ECStore runtime boundaries.
|
||||
- Current branch: `overtrue/arch-scanner-heal-runtime-boundaries`.
|
||||
- Completed slice: move scanner and heal direct ECStore runtime, disk,
|
||||
metadata, lifecycle, replication, config, and error imports behind their
|
||||
crate-local compatibility modules.
|
||||
- Acceptance: direct `rustfs_ecstore` references in `crates/scanner/src` and
|
||||
`crates/heal/src` are limited to scanner/heal compatibility boundary
|
||||
modules; scanner/heal business modules consume local compatibility names.
|
||||
- Must preserve: scanner cache load/save behavior, lifecycle and replication
|
||||
scan behavior, disk bucket scan inventory lookup, heal object/bucket/format
|
||||
behavior, resume state storage, heal channel test contracts, and existing
|
||||
ECStore-owned concrete types.
|
||||
- Risk defense: this slice changes import ownership and thin compatibility
|
||||
boundaries only; it does not alter scanner scheduling, heal scheduling,
|
||||
object I/O logic, disk operations, metadata serialization, or error
|
||||
mapping.
|
||||
- Verification: focused scanner/heal compile/tests, direct import scans,
|
||||
migration/layer guards, formatting, diff hygiene, Rust risk scan, full
|
||||
pre-commit, and required three-expert review passed.
|
||||
|
||||
## Phase 8 Background Controller Tasks
|
||||
|
||||
- [x] `BGC-001` Inventory background services.
|
||||
@@ -1114,25 +1134,26 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
|
||||
|
||||
1. `pure-move`/`consumer-migration`: continue larger cleanup slices with the
|
||||
loss-prevention guards active for remaining storage compatibility contracts
|
||||
such as scanner/heal runtime boundaries.
|
||||
around app/storage/admin runtime boundaries.
|
||||
|
||||
## Pre-Push Review Log
|
||||
|
||||
| Expert | Status | Notes |
|
||||
|---|---|---|
|
||||
| Quality/architecture | passed | Swift account, container, object, and versioning modules now use Swift-local storage compatibility wrappers for ECStore resolver and bucket metadata access. |
|
||||
| Migration preservation | passed | ECStore remains the owner of object-store resolution and bucket metadata persistence; Swift metadata tags, ACL tags, versioning tags, and object read/write/copy semantics are preserved. |
|
||||
| Testing/verification | passed | Focused Swift compile/tests, migration/layer guards, direct Swift import scan, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. |
|
||||
| Quality/architecture | passed | Scanner and heal source modules now use crate-local compatibility boundaries for ECStore runtime, disk, metadata, lifecycle, replication, config, and error imports. |
|
||||
| Migration preservation | passed | ECStore remains the owner of scanner/heal backing types and behavior; scanner cache/lifecycle/replication scans and heal object/bucket/format/resume semantics are preserved. |
|
||||
| Testing/verification | passed | Focused scanner/heal compile/tests, direct source import scans, migration/layer guards, formatting, diff hygiene, Rust risk scan, and full `make pre-commit` passed. |
|
||||
|
||||
## Verification Notes
|
||||
|
||||
Passed before push:
|
||||
|
||||
- `cargo check --tests -p rustfs-protocols`: passed.
|
||||
- `cargo test -p rustfs-protocols`: passed; 13 passed.
|
||||
- `rg -n 'rustfs_ecstore|resolve_object_store_handle|metadata_sys'
|
||||
crates/protocols/src/swift/*.rs`: remaining matches are deliberate Swift
|
||||
compatibility boundary definitions.
|
||||
- `cargo check --tests -p rustfs-scanner -p rustfs-heal`: passed.
|
||||
- `cargo test -p rustfs-scanner -p rustfs-heal`: passed.
|
||||
- `rg -n 'rustfs_ecstore' crates/scanner/src --glob '*.rs'`: remaining matches
|
||||
are deliberate scanner compatibility boundary definitions.
|
||||
- `rg -n 'rustfs_ecstore' crates/heal/src --glob '*.rs'`: remaining matches are
|
||||
deliberate heal compatibility boundary definitions.
|
||||
- `./scripts/check_architecture_migration_rules.sh`: passed.
|
||||
- `./scripts/check_layer_dependencies.sh`: passed.
|
||||
- `cargo fmt --all --check`: passed.
|
||||
@@ -1144,15 +1165,15 @@ Passed before push:
|
||||
|
||||
Notes:
|
||||
|
||||
- This slice builds on the merged API-027 compatibility cleanup.
|
||||
- Direct Swift ECStore resolver and bucket metadata access now remains only in
|
||||
the Swift compatibility boundary.
|
||||
- The slice does not alter Swift account/container metadata behavior, object
|
||||
read/write/copy behavior, versioning behavior, ACL behavior, or ECStore
|
||||
- This slice builds on the merged API-028 compatibility cleanup.
|
||||
- Direct scanner/heal ECStore imports now remain only in their compatibility
|
||||
boundary modules.
|
||||
- The slice does not alter scanner runtime behavior, heal runtime behavior,
|
||||
object I/O behavior, disk operations, metadata serialization, or ECStore
|
||||
definitions.
|
||||
|
||||
## Handoff Notes
|
||||
|
||||
- Continue with larger consumer-migration batches around scanner/heal runtime
|
||||
boundaries; keep ECStore-owned behavior in ECStore until concrete behavior is
|
||||
isolated enough for a pure-move slice.
|
||||
- Continue with larger consumer-migration batches around app/storage/admin
|
||||
runtime boundaries; keep ECStore-owned behavior in ECStore until concrete
|
||||
behavior is isolated enough for a pure-move slice.
|
||||
|
||||
Reference in New Issue
Block a user