diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index ad2241629..2d90e9762 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -575,7 +575,7 @@ pub(crate) async fn delete_confirmed_transition_candidate_exact_with_lease_idemp #[cfg(test)] static CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] pub(crate) async fn delete_confirmed_transition_candidate_exact_with_manager_and_identity( obj_name: &str, rv_id: &str, @@ -706,15 +706,16 @@ pub(crate) fn transitioned_delete_journal_entry_for_source( #[cfg(test)] mod test { + #[cfg(feature = "test-util")] + use super::delete_confirmed_transition_candidate_exact_with_manager_and_identity; use rustfs_s3_client::signer_error::invalid_utf8_header_error; use super::{ CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES, ERR_REMOTE_DELETE_BREAKER_OPEN, ERR_REMOTE_DELETE_LIMITER_CLOSED, Jentry, RemoteDeleteBreaker, RemoteTierDeleteOutcome, TierDeleteJournalState, TierDeleteSourceIdentity, - delete_confirmed_transition_candidate_exact_with_manager_and_identity, delete_object_from_remote_tier_idempotent, - delete_object_from_remote_tier_idempotent_with_manager_and_identity, is_remote_tier_not_found_error, - is_signer_header_error, lifecycle, set_remote_tier_delete_test_hook, should_record_remote_delete_failure, - transitioned_delete_journal_entry, transitioned_force_delete_journal_entry, + delete_object_from_remote_tier_idempotent, delete_object_from_remote_tier_idempotent_with_manager_and_identity, + is_remote_tier_not_found_error, is_signer_header_error, lifecycle, set_remote_tier_delete_test_hook, + should_record_remote_delete_failure, transitioned_delete_journal_entry, transitioned_force_delete_journal_entry, }; use crate::storage_api_contracts::lifecycle::TransitionedObject; use rustfs_filemeta::TransitionVersionState; diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 3c866ae6f..426956d13 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -747,7 +747,7 @@ pub enum TransitionTransactionRecoveryOutcome { OperatorRequired(IlmRecoveryErrorCode), } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] #[derive(Default)] struct TransitionRecoveryClaimBarrierState { transaction_id: Uuid, @@ -755,17 +755,17 @@ struct TransitionRecoveryClaimBarrierState { release: tokio::sync::Notify, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] pub(crate) struct TransitionRecoveryClaimBarrier { state: Arc, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] static TRANSITION_RECOVERY_CLAIM_BARRIER: std::sync::OnceLock< std::sync::Mutex>>, > = std::sync::OnceLock::new(); -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl TransitionRecoveryClaimBarrier { pub(crate) fn install(transaction_id: Uuid) -> Self { let state = Arc::new(TransitionRecoveryClaimBarrierState { @@ -796,7 +796,7 @@ impl TransitionRecoveryClaimBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl Drop for TransitionRecoveryClaimBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -810,7 +810,7 @@ impl Drop for TransitionRecoveryClaimBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] async fn pause_before_transition_recovery_claim(transaction_id: Uuid) { let barrier = TRANSITION_RECOVERY_CLAIM_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) @@ -825,7 +825,7 @@ async fn pause_before_transition_recovery_claim(transaction_id: Uuid) { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] #[derive(Default)] struct TransitionRecoveryTerminalBarrierState { transaction_id: Uuid, @@ -833,17 +833,17 @@ struct TransitionRecoveryTerminalBarrierState { release: tokio::sync::Notify, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] pub(crate) struct TransitionRecoveryTerminalBarrier { state: Arc, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] static TRANSITION_RECOVERY_TERMINAL_BARRIER: std::sync::OnceLock< std::sync::Mutex>>, > = std::sync::OnceLock::new(); -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl TransitionRecoveryTerminalBarrier { pub(crate) fn install(transaction_id: Uuid) -> Self { let state = Arc::new(TransitionRecoveryTerminalBarrierState { @@ -870,7 +870,7 @@ impl TransitionRecoveryTerminalBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl Drop for TransitionRecoveryTerminalBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -884,7 +884,7 @@ impl Drop for TransitionRecoveryTerminalBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] async fn pause_after_transition_recovery_terminal(transaction_id: Uuid) { let barrier = TRANSITION_RECOVERY_TERMINAL_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) @@ -1242,7 +1242,7 @@ async fn process_transition_transaction_record_at( }, ) .map_err(transition_transaction_store_error)?; - #[cfg(test)] + #[cfg(all(test, feature = "test-util"))] pause_before_transition_recovery_claim(current.transaction_id).await; match save_transition_transaction_record_if_current(api.clone(), ¤t, &cleanup).await { Ok(()) => recover_cleanup_pending(api.clone(), &cleanup).await, @@ -1300,7 +1300,7 @@ async fn process_transition_transaction_record_at( }; persist_transition_recovery_result(api.clone(), control, &recovery, now_unix_nanos).await?; if let Some(source) = source_to_delete { - #[cfg(test)] + #[cfg(all(test, feature = "test-util"))] pause_after_transition_recovery_terminal(source.transaction_id).await; delete_transition_transaction_record(api, &source).await?; } @@ -1322,7 +1322,7 @@ fn transition_recovery_control_identity(transaction: &TransitionTransaction, rec } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] pub(crate) fn transition_recovery_control_id(transaction: &TransitionTransaction) -> Result { let record_name = transition_transaction_record_object_name(transaction.transaction_id)?; transition_recovery_control_identity(transaction, &record_name) @@ -1753,7 +1753,7 @@ pub async fn recover_transition_transaction_records( recover_transition_transaction_records_with_now(api, limit, marker, None).await } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] pub async fn recover_transition_transaction_records_at( api: Arc, limit: usize, diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index 4032773f5..fecb3c701 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -99,11 +99,17 @@ pub(crate) const GET_STAGE_READER_OPEN_MMAP_COPY_FALLBACK: &str = "reader_open_m pub(crate) const GET_STAGE_READER_OPEN_MMAP_COPY_SUCCESS: &str = "reader_open_mmap_copy_success"; pub(crate) const GET_STAGE_READER_OPEN_STREAM: &str = "reader_open_stream"; pub(crate) const GET_STAGE_READER_MMAP_ACCESS_CHECK: &str = "reader_mmap_access_check"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_BLOCKING_TASK: &str = "reader_mmap_blocking_task"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_BLOCKING_WAIT: &str = "reader_mmap_blocking_wait"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_COPY_BUFFER: &str = "reader_mmap_copy_buffer"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_DIRECT_READ_COPY: &str = "reader_mmap_direct_read_copy"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_FILE_OPEN: &str = "reader_mmap_file_open"; +#[cfg(unix)] pub(crate) const GET_STAGE_READER_MMAP_MAP: &str = "reader_mmap_map"; pub(crate) const GET_STAGE_READER_MMAP_METADATA_LOOKUP: &str = "reader_mmap_metadata_lookup"; pub(crate) const GET_STAGE_READER_MMAP_METADATA_VALIDATE: &str = "reader_mmap_metadata_validate"; diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index b98b294ca..46487c8c9 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1301,6 +1301,7 @@ impl LocalDiskWrapper { self.disk.get_object_path(volume, path) } + #[cfg(unix)] pub(crate) fn get_object_path_for_io(&self, volume: &str, path: &str) -> crate::disk::error::Result { self.disk.get_object_path_for_io(volume, path) } diff --git a/crates/ecstore/src/disk/fs.rs b/crates/ecstore/src/disk/fs.rs index 8612b6474..7dbbaf14d 100644 --- a/crates/ecstore/src/disk/fs.rs +++ b/crates/ecstore/src/disk/fs.rs @@ -218,10 +218,12 @@ pub async fn rename(from: impl AsRef, to: impl AsRef) -> io::Result< fs::rename(from, to).await } +#[cfg(any(not(windows), test))] pub fn rename_std(from: impl AsRef, to: impl AsRef) -> io::Result<()> { std::fs::rename(from, to) } +#[cfg(any(not(windows), test))] #[tracing::instrument(level = "debug", skip_all)] pub async fn read_file(path: impl AsRef) -> io::Result> { fs::read(path.as_ref()).await diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 9cd683578..1fd76d943 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -701,7 +701,9 @@ const EVENT_DISK_LOCAL_FORMAT_DECODE_FAILED: &str = "disk_local_format_decode_fa /// to replace. Best effort — the rename that follows fails closed — but a /// recurring signal means heal is stuck on that drive. const EVENT_DISK_LOCAL_HEAL_PURGE_FAILED: &str = "disk_local_heal_purge_failed"; +#[cfg(unix)] const METRIC_GET_OBJECT_MMAP_PAGE_FAULTS_TOTAL: &str = "rustfs_io_get_object_mmap_page_faults_total"; +#[cfg(unix)] const METRIC_GET_OBJECT_DIRECT_READ_PAGE_FAULTS_TOTAL: &str = "rustfs_io_get_object_direct_read_page_faults_total"; // io_uring read-backend gray-release observability (rustfs/backlog#1172). #[cfg(target_os = "linux")] @@ -898,10 +900,15 @@ const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_ reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)" )] const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false; +#[cfg(any(unix, test))] const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE"; +#[cfg(any(unix, test))] const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false; +#[cfg(any(unix, test))] const ENV_RUSTFS_OBJECT_MMAP_READ_METHOD: &str = "RUSTFS_OBJECT_MMAP_READ_METHOD"; +#[cfg(any(unix, test))] const RUSTFS_OBJECT_MMAP_READ_METHOD_MMAP_COPY: &str = "mmap_copy"; +#[cfg(any(unix, test))] const RUSTFS_OBJECT_MMAP_READ_METHOD_DIRECT_READ_COPY: &str = "direct_read_copy"; /// Legacy binary switch for commit-point durability (fsync writes and renames). @@ -917,6 +924,7 @@ const DEFAULT_RUSTFS_DRIVE_SYNC_ENABLE: bool = true; /// See docs/operations/durability-modes.md for the power-loss guarantee matrix. const ENV_RUSTFS_DURABILITY_MODE: &str = "RUSTFS_DURABILITY_MODE"; +#[cfg(any(unix, test))] #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum LocalReadCopyMethod { MmapCopy, @@ -1325,15 +1333,18 @@ cached_read_env! { cached_read_env! { /// Whether mmap reads should fault the mapping in with `MAP_POPULATE`. + #[cfg(any(unix, test))] fn mmap_populate_enabled() -> bool = rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE, DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE); } +#[cfg(any(unix, test))] fn should_populate_mmap_read(length: usize) -> bool { length > 0 && mmap_populate_enabled() } cached_read_env! { + #[cfg(any(unix, test))] fn local_read_copy_method() -> LocalReadCopyMethod = { let method = rustfs_utils::get_env_str(ENV_RUSTFS_OBJECT_MMAP_READ_METHOD, RUSTFS_OBJECT_MMAP_READ_METHOD_MMAP_COPY); match method.as_str() { @@ -1976,7 +1987,7 @@ fn set_inline_preparation_before_backup(dst_path: &str, hook: impl FnOnce() + Se .insert(dst_path.to_string(), Box::new(hook)); } -#[cfg(test)] +#[cfg(all(test, unix))] fn set_inline_before_file_sync_admission(dst_path: &str, hook: impl FnOnce() + Send + 'static) { INLINE_BEFORE_FILE_SYNC_ADMISSION .lock() @@ -2223,7 +2234,7 @@ fn should_remove_staged_meta_before_commit(_dst_path: &str) -> bool { false } -#[cfg(not(test))] +#[cfg(all(not(test), not(windows)))] fn should_fail_local_inline_rollback_hardlink(_dst_path: &Path) -> bool { false } @@ -3724,6 +3735,7 @@ struct FdKey { /// The generation fence and explicit mutation invalidation keep the snapshot /// tied to the inode held by `file`, allowing cache hits to avoid a repeated /// metadata syscall without weakening replacement/heal semantics. +#[cfg(unix)] struct FdCacheEntry { /// An independently cloneable descriptor for the immutable shard inode. file: Arc, @@ -5474,6 +5486,7 @@ impl LocalDisk { local_disk_bucket_path(&self.root, bucket) } + #[cfg(any(unix, test))] pub(crate) fn get_object_path_for_io(&self, bucket: &str, key: &str) -> Result { self.io_get_object_path(bucket, key) } @@ -18709,11 +18722,17 @@ mod test { path_resolve_stage: "path", metadata_lookup_stage: "metadata_lookup", metadata_validate_stage: "metadata_validate", + #[cfg(unix)] blocking_wait_stage: "blocking_wait", + #[cfg(unix)] blocking_task_stage: "blocking_task", + #[cfg(unix)] file_open_stage: "file_open", + #[cfg(unix)] mmap_map_stage: "mmap_map", + #[cfg(unix)] mmap_copy_stage: "mmap_copy", + #[cfg(unix)] direct_read_copy_stage: "direct_read_copy", }; diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index 9d51bc1cd..94c57d313 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -17,13 +17,14 @@ #[cfg(all(test, windows))] use super::run_destination_commit_directory_preparation; +#[cfg(any(not(windows), test))] +use super::should_fail_local_inline_rollback_hardlink; use super::{ EVENT_DISK_LOCAL_ACCESS_FAILED, EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, EVENT_DISK_LOCAL_RENAME_REJECTED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_DISK_LOCAL, LocalDisk, SyncMode, effective_durability, inline_metadata_rollback_dir, observe_old_current_size, remove_dir_all_if_exists, remove_dst_base_before_commit, remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup, should_fail_after_metadata_commit, should_fail_before_old_metadata_backup, - should_fail_commit_rename, should_fail_local_inline_rollback_hardlink, should_remove_staged_meta_before_commit, - skip_access_checks, + should_fail_commit_rename, should_remove_staged_meta_before_commit, skip_access_checks, }; #[cfg(test)] use super::{run_inline_before_file_sync_admission, run_owned_file_write_before_open, run_rename_data_after_first_publication}; @@ -86,6 +87,7 @@ fn rollback_inline_metadata_commit_std( Ok(()) } +#[cfg(any(not(windows), test))] pub(super) fn create_local_inline_rollback_backup( dst_file_path: &Path, staging_file_path: &Path, diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index c2f2c52b4..aac24b781 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -190,11 +190,17 @@ pub struct MmapCopyStageMetrics { pub(crate) path_resolve_stage: &'static str, pub(crate) metadata_lookup_stage: &'static str, pub(crate) metadata_validate_stage: &'static str, + #[cfg(unix)] pub(crate) blocking_wait_stage: &'static str, + #[cfg(unix)] pub(crate) blocking_task_stage: &'static str, + #[cfg(unix)] pub(crate) file_open_stage: &'static str, + #[cfg(unix)] pub(crate) mmap_map_stage: &'static str, + #[cfg(unix)] pub(crate) mmap_copy_stage: &'static str, + #[cfg(unix)] pub(crate) direct_read_copy_stage: &'static str, } @@ -922,6 +928,7 @@ impl Disk { } } + #[cfg(unix)] pub(crate) fn get_object_path_for_io_if_local( &self, volume: &str, diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index e68a51d99..a0f5e0580 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -91,6 +91,7 @@ pub(crate) mod fsync_dir_recorder { static RECORDED: Mutex> = Mutex::new(Vec::new()); static LIMITED: Mutex> = Mutex::new(Vec::new()); static GROUPED: Mutex> = Mutex::new(Vec::new()); + #[cfg(unix)] static BEFORE_LIMITED: std::sync::LazyLock>> = std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); static BEFORE_GROUP_BATCH: std::sync::LazyLock>> = @@ -150,6 +151,7 @@ pub(crate) mod fsync_dir_recorder { contains_path(&RECORDED.lock().expect("fsync dir recorder poisoned"), dir) } + #[cfg(unix)] pub(crate) fn record_limited(dir: &Path) { record_path(&LIMITED, dir, "limited fsync dir recorder"); let hook = remove_hook(&BEFORE_LIMITED, dir, "limited fsync hook poisoned"); @@ -162,6 +164,7 @@ pub(crate) mod fsync_dir_recorder { contains_path(&LIMITED.lock().expect("limited fsync dir recorder poisoned"), dir) } + #[cfg(unix)] pub(crate) fn set_before_limited(dir: &Path, hook: impl FnOnce() + Send + 'static) { BEFORE_LIMITED .lock() @@ -237,6 +240,7 @@ pub(crate) mod fsync_dir_recorder { .insert(dir.to_path_buf(), kind); } + #[cfg(unix)] pub(crate) fn take_grouped_failure(dir: &Path) -> Option { remove_path_keyed(&GROUPED_FAILURES, dir, "grouped fsync failure hook poisoned") } diff --git a/crates/ecstore/src/io_support/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs index 371a93560..d5d5f06de 100644 --- a/crates/ecstore/src/io_support/bitrot.rs +++ b/crates/ecstore/src/io_support/bitrot.rs @@ -13,12 +13,15 @@ // limitations under the License. use crate::diagnostics::get::{ - GET_STAGE_READER_MMAP_ACCESS_CHECK, GET_STAGE_READER_MMAP_BLOCKING_TASK, GET_STAGE_READER_MMAP_BLOCKING_WAIT, - GET_STAGE_READER_MMAP_COPY_BUFFER, GET_STAGE_READER_MMAP_DIRECT_READ_COPY, GET_STAGE_READER_MMAP_FILE_OPEN, - GET_STAGE_READER_MMAP_MAP, GET_STAGE_READER_MMAP_METADATA_LOOKUP, GET_STAGE_READER_MMAP_METADATA_VALIDATE, + GET_STAGE_READER_MMAP_ACCESS_CHECK, GET_STAGE_READER_MMAP_METADATA_LOOKUP, GET_STAGE_READER_MMAP_METADATA_VALIDATE, GET_STAGE_READER_MMAP_PATH_RESOLVE, GET_STAGE_READER_OPEN_MMAP_COPY_FALLBACK, GET_STAGE_READER_OPEN_MMAP_COPY_SUCCESS, GET_STAGE_READER_OPEN_STREAM, GET_STAGE_READER_STREAM_FIRST_READ, record_get_stage_duration_if_enabled, }; +#[cfg(unix)] +use crate::diagnostics::get::{ + GET_STAGE_READER_MMAP_BLOCKING_TASK, GET_STAGE_READER_MMAP_BLOCKING_WAIT, GET_STAGE_READER_MMAP_COPY_BUFFER, + GET_STAGE_READER_MMAP_DIRECT_READ_COPY, GET_STAGE_READER_MMAP_FILE_OPEN, GET_STAGE_READER_MMAP_MAP, +}; #[cfg(feature = "hotpath")] use crate::disk::FileWriter; use crate::disk::{self, DiskAPI as _, DiskStore, FileReader, MmapCopyStageMetrics, error::DiskError}; @@ -406,11 +409,17 @@ async fn open_disk_reader( path_resolve_stage: GET_STAGE_READER_MMAP_PATH_RESOLVE, metadata_lookup_stage: GET_STAGE_READER_MMAP_METADATA_LOOKUP, metadata_validate_stage: GET_STAGE_READER_MMAP_METADATA_VALIDATE, + #[cfg(unix)] blocking_wait_stage: GET_STAGE_READER_MMAP_BLOCKING_WAIT, + #[cfg(unix)] blocking_task_stage: GET_STAGE_READER_MMAP_BLOCKING_TASK, + #[cfg(unix)] file_open_stage: GET_STAGE_READER_MMAP_FILE_OPEN, + #[cfg(unix)] mmap_map_stage: GET_STAGE_READER_MMAP_MAP, + #[cfg(unix)] mmap_copy_stage: GET_STAGE_READER_MMAP_COPY_BUFFER, + #[cfg(unix)] direct_read_copy_stage: GET_STAGE_READER_MMAP_DIRECT_READ_COPY, }); let mmap_result = { diff --git a/crates/ecstore/src/services/tier/test_util.rs b/crates/ecstore/src/services/tier/test_util.rs index 59b4d6f4e..09c6dee81 100644 --- a/crates/ecstore/src/services/tier/test_util.rs +++ b/crates/ecstore/src/services/tier/test_util.rs @@ -56,6 +56,7 @@ use std::collections::HashMap; use std::io::Cursor; +#[cfg(feature = "test-util")] use std::path::Path; use std::sync::{ Arc, @@ -68,21 +69,28 @@ use tokio::io::AsyncReadExt; use tokio::sync::{Mutex, Notify, RwLock}; use uuid::Uuid; +#[cfg(feature = "test-util")] use crate::disk::endpoint::Endpoint; +#[cfg(feature = "test-util")] use crate::disk::format::FormatV3; +#[cfg(feature = "test-util")] use crate::disk::{DiskAPI, DiskOption, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE, new_disk}; use crate::services::tier::tier::TierConfigMgr; use crate::services::tier::tier_config::{TierConfig, TierMinIO, TierType}; use crate::services::tier::warm_backend::{ TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options, }; +#[cfg(feature = "test-util")] use rustfs_filemeta::FileMeta; use rustfs_s3_client::transition_api::{ReadCloser, ReaderImpl}; +#[cfg(feature = "test-util")] use rustfs_utils::path::path_join_buf; /// One-shot barrier before rejected transition cleanup resolves its ECStore. +#[cfg(feature = "test-util")] pub struct TransitionCleanupStoreBarrier(crate::set_disk::SetDiskTransitionCleanupStoreBarrier); +#[cfg(feature = "test-util")] impl TransitionCleanupStoreBarrier { /// Install the barrier for the next rejected transition cleanup. pub fn install() -> Self { @@ -96,6 +104,7 @@ impl TransitionCleanupStoreBarrier { } /// Default polling cadence used by the `wait_for_*` helpers. +#[cfg(feature = "test-util")] const POLL_INTERVAL: Duration = Duration::from_millis(50); /// A fault to inject into [`MockWarmBackend`] operations. @@ -208,10 +217,12 @@ impl Drop for MockRemoveOperationGuard { } /// One-shot barrier that pauses a mock tier PUT after storing its remote body. +#[cfg(feature = "test-util")] pub struct MockPutBarrier { state: Arc, } +#[cfg(feature = "test-util")] impl MockPutBarrier { /// Wait until the remote body is stored and the PUT is paused before returning. pub async fn wait_until_paused(&self) { @@ -226,6 +237,7 @@ impl MockPutBarrier { } } +#[cfg(feature = "test-util")] impl Drop for MockPutBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -258,10 +270,12 @@ impl Drop for MockGetBarrier { } /// One-shot barrier that pauses and then fails a mock tier DELETE. +#[cfg(feature = "test-util")] pub struct MockRemoveBarrier { state: Arc, } +#[cfg(feature = "test-util")] impl MockRemoveBarrier { /// Wait until DELETE reaches the deterministic failure point. pub async fn wait_until_paused(&self) { @@ -283,6 +297,7 @@ impl MockRemoveBarrier { } } +#[cfg(feature = "test-util")] impl Drop for MockRemoveBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -306,6 +321,7 @@ impl MockWarmBackend { } /// Arm a one-shot pause after the next tier PUT stores its remote body. + #[cfg(feature = "test-util")] pub async fn arm_put_barrier(&self) -> MockPutBarrier { let state = Arc::new(MockPutBarrierState::default()); *self.inner.put_barrier.lock().await = Some(Arc::clone(&state)); @@ -313,6 +329,7 @@ impl MockWarmBackend { } /// Pause and then fail the next DELETE after it reaches the backend. + #[cfg(feature = "test-util")] pub async fn arm_failing_remove_barrier(&self) -> MockRemoveBarrier { let state = Arc::new(MockRemoveBarrierState::default()); let mut barrier = self.inner.remove_barrier.lock().await; @@ -323,6 +340,7 @@ impl MockWarmBackend { /// Arm a one-shot pause before the next tier GET, then return an error /// after the test releases it. + #[cfg(feature = "test-util")] pub async fn arm_failing_get_barrier(&self) -> MockGetBarrier { let state = Arc::new(MockGetBarrierState { fail_after_release: true, @@ -343,6 +361,7 @@ impl MockWarmBackend { // ---- fault injection ------------------------------------------------- /// Replace the entire fault configuration. + #[cfg(feature = "test-util")] pub async fn set_faults(&self, faults: FaultConfig) { *self.inner.faults.lock().await = faults; } @@ -353,6 +372,7 @@ impl MockWarmBackend { } /// Toggle "HTTP 5xx" server errors on every operation. + #[cfg(feature = "test-util")] pub async fn set_server_error(&self, server_error: bool) { self.inner.faults.lock().await.server_error = server_error; } @@ -363,11 +383,13 @@ impl MockWarmBackend { } /// Set (or clear, with `None`) injected latency applied before each op. + #[cfg(feature = "test-util")] pub async fn set_latency(&self, latency: Option) { self.inner.faults.lock().await.latency = latency; } /// Clear all injected faults, restoring healthy behaviour. + #[cfg(feature = "test-util")] pub async fn clear_faults(&self) { *self.inner.faults.lock().await = FaultConfig::default(); } @@ -375,6 +397,7 @@ impl MockWarmBackend { /// Limit how many body bytes a successful mock PUT consumes. `None` drains /// the complete body. This models a backend that incorrectly accepts a /// truncated stream while still returning success. + #[cfg(feature = "test-util")] pub async fn set_put_read_limit(&self, limit: Option) { *self.inner.put_read_limit.lock().await = limit; } @@ -395,12 +418,14 @@ impl MockWarmBackend { } /// Reject non-empty remote versions before transition metadata is committed. + #[cfg(feature = "test-util")] pub fn set_reject_non_empty_remote_versions(&self, reject: bool) { self.inner.reject_non_empty_remote_versions.store(reject, Ordering::Release); } /// Reject the next non-empty remote version validation without changing /// subsequent exact-version backend cleanup behavior. + #[cfg(feature = "test-util")] pub fn reject_next_non_empty_remote_version_validation(&self) { self.inner .reject_non_empty_remote_version_validations @@ -438,6 +463,7 @@ impl MockWarmBackend { } /// Clear the operation log without touching stored objects or faults. + #[cfg(feature = "test-util")] pub async fn clear_op_log(&self) { self.inner.op_log.lock().await.clear(); } @@ -459,11 +485,13 @@ impl MockWarmBackend { } /// Return the exact object/version pairs produced by successful tier PUTs. + #[cfg(feature = "test-util")] pub async fn put_versions(&self) -> Vec<(String, String)> { self.inner.put_versions.lock().await.clone() } /// Return the exact object/version pairs passed to successful tier removes. + #[cfg(feature = "test-util")] pub async fn remove_versions(&self) -> Vec<(String, String)> { self.inner.remove_versions.lock().await.clone() } @@ -475,6 +503,7 @@ impl MockWarmBackend { /// Number of `get` calls recorded — useful to assert restore reads hit the /// local copy rather than the remote tier. + #[cfg(feature = "test-util")] pub async fn get_count(&self) -> usize { self.inner .op_log @@ -486,6 +515,7 @@ impl MockWarmBackend { } /// Number of `put` calls recorded. + #[cfg(feature = "test-util")] pub async fn put_count(&self) -> usize { self.inner .op_log @@ -499,6 +529,7 @@ impl MockWarmBackend { // ---- storage inspection --------------------------------------------- /// Whether the backend currently stores `object`. + #[cfg(feature = "test-util")] pub async fn contains(&self, object: &str) -> bool { self.inner.objects.lock().await.contains_key(object) } @@ -509,11 +540,13 @@ impl MockWarmBackend { } /// A clone of the stored object, if present. + #[cfg(feature = "test-util")] pub async fn stored(&self, object: &str) -> Option { self.inner.objects.lock().await.get(object).cloned() } /// A clone of the raw bytes stored for `object`, if present. + #[cfg(feature = "test-util")] pub async fn bytes(&self, object: &str) -> Option> { self.inner.objects.lock().await.get(object).map(|o| o.bytes.clone()) } @@ -538,6 +571,7 @@ impl MockWarmBackend { /// Poll until `object` is absent from the backend, or `timeout` elapses. /// Returns `true` if the object disappeared within the budget. + #[cfg(feature = "test-util")] pub async fn wait_for_remote_absence(&self, object: &str, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; loop { @@ -553,6 +587,7 @@ impl MockWarmBackend { /// Poll until the backend holds exactly `expected` objects, or `timeout` /// elapses. Returns `true` if the count was reached within the budget. + #[cfg(feature = "test-util")] pub async fn wait_for_object_count(&self, expected: usize, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; loop { @@ -847,6 +882,7 @@ pub async fn register_mock_tier_backend(handle: &Arc>, tie /// The transition-state tuple read from an on-disk `xl.meta`, plus the object's /// free-version count. #[derive(Clone, Debug, PartialEq, Eq)] +#[cfg(feature = "test-util")] pub struct TransitionMeta { /// `transition_status` (e.g. `"complete"`), empty when not transitioned. pub status: String, @@ -860,6 +896,7 @@ pub struct TransitionMeta { pub free_version_count: usize, } +#[cfg(feature = "test-util")] async fn open_disk(disk_path: &Path) -> Option { // `LocalDisk::new` rejects an endpoint whose (set_idx, disk_idx) disagrees // with the position recorded in the disk's own format.json, so derive the @@ -890,6 +927,7 @@ async fn open_disk(disk_path: &Path) -> Option { /// The free-version metadata removal lands asynchronously after the remote /// object disappears, so callers typically poll via /// [`wait_for_free_version_absence`] instead of asserting a single read. +#[cfg(feature = "test-util")] pub async fn free_version_count(disk_path: &Path, bucket: &str, object: &str) -> usize { let Some(disk) = open_disk(disk_path).await else { return 0; @@ -914,6 +952,7 @@ pub async fn free_version_count(disk_path: &Path, bucket: &str, object: &str) -> /// fields are taken from the newest version that carries a transition record; /// if no version is transitioned, they are taken from the current version (and /// will be empty). +#[cfg(feature = "test-util")] pub async fn read_transition_meta(disk_path: &Path, bucket: &str, object: &str) -> Option { let disk = open_disk(disk_path).await?; let data = disk @@ -947,6 +986,7 @@ pub async fn read_transition_meta(disk_path: &Path, bucket: &str, object: &str) /// disk is missing the object or disagrees — this is the shard-consistency /// check required by ilm-6 (the `(status, tier, remote key, remote version id)` /// four-tuple plus free-version count must match across all erasure shards). +#[cfg(feature = "test-util")] pub async fn assert_transition_meta_consistent>(disk_paths: &[P], bucket: &str, object: &str) -> TransitionMeta { assert!(!disk_paths.is_empty(), "assert_transition_meta_consistent needs at least one disk"); @@ -972,6 +1012,7 @@ pub async fn assert_transition_meta_consistent>(disk_paths: &[P], /// Poll until `object` retains no free versions on `disk_path`, or `timeout` /// elapses. Returns `true` if the free versions drained within the budget. +#[cfg(feature = "test-util")] pub async fn wait_for_free_version_absence(disk_path: &Path, bucket: &str, object: &str, timeout: Duration) -> bool { let deadline = tokio::time::Instant::now() + timeout; loop { @@ -1040,6 +1081,44 @@ mod tests { ); } + #[tokio::test] + async fn mock_metadata_survives_put_and_external_delete_is_distinct() { + let backend = MockWarmBackend::new(); + let metadata = HashMap::from([ + ("content-type".to_string(), "text/plain".to_string()), + ("project".to_string(), "archive".to_string()), + ]); + let version = backend + .put_with_meta("object", ReaderImpl::Body(Bytes::from_static(b"body")), 4, metadata.clone()) + .await + .expect("mock PUT should preserve remote metadata"); + assert_eq!(backend.metadata("object").await, Some(metadata)); + assert_eq!( + backend + .probe_transition_candidate_state("object") + .await + .expect("probe stored object"), + TransitionCandidateProbe::VersionedPresent(version) + ); + + backend.external_remove("object").await; + assert_eq!(backend.metadata("object").await, None); + assert_eq!( + backend + .probe_transition_candidate_state("object") + .await + .expect("probe removed object"), + TransitionCandidateProbe::Missing + ); + let operations = backend.op_log().await; + assert!( + operations + .iter() + .any(|op| matches!(op, MockWarmOp::ExternalRemove { object } if object == "object")) + ); + assert!(!operations.iter().any(|op| matches!(op, MockWarmOp::Remove { .. }))); + } + #[tokio::test] async fn mock_probe_preserves_fault_fail_closed_behavior() { let backend = MockWarmBackend::new(); diff --git a/crates/ecstore/src/set_disk/core/io_primitives.rs b/crates/ecstore/src/set_disk/core/io_primitives.rs index 14da6fc97..5c0b16a1f 100644 --- a/crates/ecstore/src/set_disk/core/io_primitives.rs +++ b/crates/ecstore/src/set_disk/core/io_primitives.rs @@ -111,10 +111,11 @@ use crate::disk::{ use crate::erasure::coding::BitrotReader; use crate::io_support::bitrot::ShardReader; use crate::io_support::bitrot::{ - BitrotReaderStageMetrics, DeferredReaderStripeHandle, adjust_shard_read_params, - create_bitrot_reader_from_bytes_with_stage_metrics, create_deferred_bitrot_reader_with_stripe_handle, - object_mmap_read_max_length, + BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_from_bytes_with_stage_metrics, + create_deferred_bitrot_reader_with_stripe_handle, }; +#[cfg(unix)] +use crate::io_support::bitrot::{adjust_shard_read_params, object_mmap_read_max_length}; use crate::set_disk::runtime_sources; use crate::set_disk::shard_source::ShardReadCost; use crate::storage_api_contracts::object::ObjectOperations; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 2c076efea..179c4fd38 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -874,7 +874,7 @@ pub(crate) use ops::multipart::NewMultipartUploadCommitObservation; pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause}; #[cfg(test)] pub(crate) use ops::object::DeleteObjectCommitBarrier; -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier; #[cfg(all(test, feature = "test-util"))] pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransitionUploadedCommitBarrier; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 3e09515d2..c57f0b36b 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -5785,7 +5785,7 @@ pub(crate) async fn cleanup_rejected_transition_upload_durably( } async fn transition_cleanup_store(ctx: &Arc) -> Option> { - #[cfg(any(test, feature = "test-util"))] + #[cfg(feature = "test-util")] pause_transition_cleanup_store().await; transition_object_store(ctx).await @@ -6031,24 +6031,24 @@ async fn delete_transition_transaction_after_remote_cleanup( } } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] #[derive(Default)] struct TransitionCleanupStoreBarrierState { arrived: tokio::sync::Notify, release: tokio::sync::Notify, } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] /// One-shot test barrier placed before transition cleanup resolves its ECStore. pub(crate) struct TransitionCleanupStoreBarrier { state: Arc, } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] static TRANSITION_CLEANUP_STORE_BARRIER: std::sync::OnceLock>>> = std::sync::OnceLock::new(); -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] impl TransitionCleanupStoreBarrier { /// Install the process-local barrier for the next cleanup-store resolution. pub(crate) fn install() -> Self { @@ -6071,7 +6071,7 @@ impl TransitionCleanupStoreBarrier { } } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] impl Drop for TransitionCleanupStoreBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -6085,7 +6085,7 @@ impl Drop for TransitionCleanupStoreBarrier { } } -#[cfg(any(test, feature = "test-util"))] +#[cfg(feature = "test-util")] async fn pause_transition_cleanup_store() { let barrier = TRANSITION_CLEANUP_STORE_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) @@ -6159,7 +6159,7 @@ async fn pause_after_transition_upload_candidate_recorded() { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] struct TransitionUploadedCommitBarrierState { bucket: String, object: String, @@ -6167,17 +6167,17 @@ struct TransitionUploadedCommitBarrierState { release: tokio::sync::Notify, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] pub(crate) struct TransitionUploadedCommitBarrier { state: Arc, } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] static TRANSITION_UPLOADED_COMMIT_BARRIER: std::sync::OnceLock< std::sync::Mutex>>, > = std::sync::OnceLock::new(); -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl TransitionUploadedCommitBarrier { pub(crate) fn install(bucket: &str, object: &str) -> Self { let state = Arc::new(TransitionUploadedCommitBarrierState { @@ -6210,7 +6210,7 @@ impl TransitionUploadedCommitBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] impl Drop for TransitionUploadedCommitBarrier { fn drop(&mut self) { self.state.release.notify_one(); @@ -6224,7 +6224,7 @@ impl Drop for TransitionUploadedCommitBarrier { } } -#[cfg(test)] +#[cfg(all(test, feature = "test-util"))] async fn pause_after_transition_uploaded_persisted(bucket: &str, object: &str) { let barrier = TRANSITION_UPLOADED_COMMIT_BARRIER .get_or_init(|| std::sync::Mutex::new(None)) @@ -9154,7 +9154,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } upload_cleanup.update_cleanup_transaction(&transaction); - #[cfg(test)] + #[cfg(all(test, feature = "test-util"))] pause_after_transition_uploaded_persisted(bucket, object).await; let commit_opts = opts.as_commit_opts(); diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 908572998..469a5e365 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -238,7 +238,7 @@ async fn list_pool_multipart_uploads_for_incarnation( } impl ECStore { - #[cfg(test)] + #[cfg(all(test, feature = "test-util"))] pub(crate) fn reset_data_movement_multipart_discovery_count_for_test(&self) { data_movement_multipart_discovery_counts() .lock() @@ -246,7 +246,7 @@ impl ECStore { .insert(self.id, 0); } - #[cfg(test)] + #[cfg(all(test, feature = "test-util"))] pub(crate) fn data_movement_multipart_discovery_count_for_test(&self) -> usize { data_movement_multipart_discovery_counts() .lock()