diff --git a/crates/config/src/constants/object.rs b/crates/config/src/constants/object.rs index 62180919c..36dd9dbf3 100644 --- a/crates/config/src/constants/object.rs +++ b/crates/config/src/constants/object.rs @@ -67,6 +67,21 @@ pub const DEFAULT_OBJECT_MEDIUM_CONCURRENCY_THRESHOLD: usize = 4; /// Default is set to 64 concurrent reads. pub const DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS: usize = 64; +/// Environment variable for the disk read permit wait timeout (seconds). +/// - Purpose: Bound how long a GET waits for a disk read permit before proceeding without one. +/// - Unit: seconds (u64). `0` waits indefinitely. +/// - Example: `export RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT=5` +pub const ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: &str = "RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT"; + +/// Maximum time a GET request waits for a disk read permit (seconds). +/// +/// Permits are held for the whole response body transfer, so slow clients can +/// occupy all of them while the disks sit idle. Instead of stalling until the +/// request-level timeout fires, a GET that waits longer than this proceeds +/// without a permit (degraded pass-through) and the bypass is counted in +/// metrics/logs. Set to 0 to wait indefinitely (previous behavior). +pub const DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT: u64 = 5; + /// Skip bitrot hash verification on GetObject reads. /// /// When enabled, GetObject reads skip the per-shard hash diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 67b5e44dc..dcea2d879 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -86,8 +86,8 @@ pub mod disk { pub use crate::disk::{ BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore, - FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, - RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, + FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, + ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, validate_batch_read_version_item_count, }; pub use crate::disk::{endpoint, error, error_reduce}; @@ -145,8 +145,8 @@ pub mod notification { pub mod object { pub use crate::object_api::{ - BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, RangedDecompressReader, - StreamConsumer, + BLOCK_SIZE_V2, ERASURE_ALGORITHM, GetObjectBodyCacheHook, GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, + RangedDecompressReader, StreamConsumer, register_get_object_body_cache_hook, }; } diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index 1872e9bef..7c896eed0 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -22,6 +22,7 @@ pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE: &str = "codec_st pub(crate) const GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE: &str = "codec_streaming_rustfs_engine"; pub(crate) const GET_OBJECT_PATH_EMPTY: &str = "empty"; pub(crate) const GET_OBJECT_PATH_DIRECT_MEMORY: &str = "direct_memory"; +pub(crate) const GET_OBJECT_PATH_BODY_CACHE: &str = "body_cache"; pub(crate) const GET_OBJECT_PATH_INLINE_DIRECT: &str = "inline_direct"; pub(crate) const GET_OBJECT_PATH_LEGACY_DUPLEX: &str = "legacy_duplex"; pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition"; diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 56b0906f1..42b42e72f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -864,6 +864,14 @@ impl LocalDisk { Vec::new() } }; + // An erasure-set heal drops a marker on the disks it is + // rebuilding (see rustfs-heal); surface it so scanner + // coordination, lock selection and admin/metrics see + // the rebuild. Refreshed with this cache (~1s). + let healing = + tokio::fs::try_exists(root.join(super::RUSTFS_META_BUCKET).join(super::HEALING_MARKER_PATH)) + .await + .unwrap_or(false); let disk_info = DiskInfo { total: info.total, free: info.free, @@ -876,13 +884,13 @@ impl LocalDisk { root_disk: is_root_disk, physical_device_ids, id: disk_id, + healing, ..Default::default() }; // if root { // return Err(Error::new(DiskError::DriveIsRoot)); // } - // disk_info.healing = Ok(disk_info) } Err(err) => Err(err.into()), diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 6c09fe848..225fafa90 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -33,6 +33,9 @@ pub const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp"; pub const RUSTFS_META_TMP_DELETED_BUCKET: &str = ".rustfs.sys/tmp/.trash"; pub const BUCKET_META_PREFIX: &str = "buckets"; pub const FORMAT_CONFIG_FILE: &str = "format.json"; +/// Per-disk marker present while an erasure-set heal is rebuilding this disk. +/// `LocalDisk::disk_info` reports `healing = true` while the file exists. +pub const HEALING_MARKER_PATH: &str = "healing.bin"; pub const STORAGE_FORMAT_FILE: &str = "xl.meta"; pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp"; diff --git a/crates/ecstore/src/object_api/body_cache_hook.rs b/crates/ecstore/src/object_api/body_cache_hook.rs new file mode 100644 index 000000000..14fd1e7e7 --- /dev/null +++ b/crates/ecstore/src/object_api/body_cache_hook.rs @@ -0,0 +1,49 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Hook that lets the application layer serve a GET body from its object data +//! cache after metadata resolution but before the erasure data read. +//! +//! The cache itself lives above ecstore (it needs app-level config, metrics +//! and invalidation), but the lookup must happen inside `get_object_reader`: +//! probing earlier would require a second metadata fan-out, and probing later +//! (after the reader is built) means a hit no longer saves any disk I/O. + +use crate::object_api::ObjectInfo; +use bytes::Bytes; +use std::sync::{Arc, OnceLock}; + +/// Serves full-object GET bodies from a cache keyed by object identity. +/// +/// Implementations must validate identity (etag/version/size) against the +/// provided `ObjectInfo`, which reflects the just-resolved metadata quorum, +/// and must return `None` for anything they cannot serve byte-identically +/// (encrypted objects, remote/transitioned objects, size mismatches, ...). +#[async_trait::async_trait] +pub trait GetObjectBodyCacheHook: Send + Sync + 'static { + async fn lookup(&self, bucket: &str, object: &str, info: &ObjectInfo) -> Option; +} + +static GET_OBJECT_BODY_CACHE_HOOK: OnceLock> = OnceLock::new(); + +/// Register the process-wide GET body cache hook. First registration wins; +/// later calls are ignored so tests and re-inits cannot swap the hook midway. +pub fn register_get_object_body_cache_hook(hook: Arc) { + let _ = GET_OBJECT_BODY_CACHE_HOOK.set(hook); +} + +/// The registered hook, if any. +pub(crate) fn get_object_body_cache_hook() -> Option<&'static Arc> { + GET_OBJECT_BODY_CACHE_HOOK.get() +} diff --git a/crates/ecstore/src/object_api/mod.rs b/crates/ecstore/src/object_api/mod.rs index 12a180121..7b2837858 100644 --- a/crates/ecstore/src/object_api/mod.rs +++ b/crates/ecstore/src/object_api/mod.rs @@ -52,8 +52,11 @@ use uuid::Uuid; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; pub const BLOCK_SIZE_V2: usize = 1024 * 1024; // 1M +mod body_cache_hook; mod readers; mod types; +pub(crate) use body_cache_hook::get_object_body_cache_hook; +pub use body_cache_hook::{GetObjectBodyCacheHook, register_get_object_body_cache_hook}; pub use readers::*; pub use types::*; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4b627c132..b58e74448 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -27,7 +27,7 @@ use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl}; use crate::cluster::rpc::heal_bucket_local_on_disks; use crate::data_usage::record_compression_total_memory; use crate::diagnostics::get::{ - GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_CODEC_STREAMING, + GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART, GET_OBJECT_PATH_BODY_CACHE, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, GET_OBJECT_PATH_DIRECT_MEMORY, GET_OBJECT_PATH_EMPTY, GET_OBJECT_PATH_INLINE_DIRECT, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_OBJECT_PATH_REMOTE_TRANSITION, GET_OBJECT_PATH_SET_DISK, GET_STAGE_DECODE, GET_STAGE_EMIT, GET_STAGE_INLINE_PREPARE, GET_STAGE_LOCK_ACQUIRE, @@ -51,6 +51,7 @@ use crate::error::{Error, Result, is_err_version_not_found}; use crate::error::{GenericError, ObjectApiError, is_err_object_not_found}; use crate::io_support::bitrot::{create_bitrot_reader, create_bitrot_reader_from_bytes, create_bitrot_writer}; use crate::object_api::ObjectOptions; +use crate::object_api::get_object_body_cache_hook; use crate::runtime::sources as runtime_sources; use crate::services::batch_processor::AsyncBatchProcessor; use crate::storage_api_contracts::{ @@ -453,11 +454,27 @@ mod write; /// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds) /// Defaults to 30 seconds if not set or invalid +/// Lock acquisition timeout. Cached: this is consulted on every object +/// lock acquisition and `std::env::var` takes a process-global lock. In test +/// builds the env var is read directly so `temp_env` overrides take effect. pub fn get_lock_acquire_timeout() -> Duration { - Duration::from_secs(rustfs_utils::get_env_u64( - rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, - rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, - )) + #[cfg(test)] + { + Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, + rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, + )) + } + #[cfg(not(test))] + { + static CACHED: OnceLock = OnceLock::new(); + *CACHED.get_or_init(|| { + Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, + rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT, + )) + }) + } } pub fn is_object_lock_diag_enabled() -> bool { @@ -2332,6 +2349,27 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { )); } + // App-layer object data cache probe: metadata (etag/size) is resolved + // but no data shards have been read yet, so a hit skips the erasure + // read, bitrot verify and decode entirely. The hook validates object + // identity and rejects anything it cannot serve byte-identically. + if range.is_none() + && opts.part_number.is_none() + && let Some(hook) = get_object_body_cache_hook() + && let Some(body) = hook.lookup(bucket, object, &object_info).await + { + record_get_object_reader_path_observation(GET_OBJECT_PATH_BODY_CACHE, object_class, size_bucket); + let reader = GetObjectReader { + stream: Box::new(Cursor::new(body.clone())), + object_info, + buffered_body: Some(body), + }; + if lock_optimization_enabled { + release_materialized_read_lock(bucket, object, read_lock_guard.take()); + } + return Ok(reader); + } + if is_get_small_object_direct_memory_eligible(&range, &object_info, &fi, opts) { let object_size = usize::try_from(object_info.size) .map_err(|_| to_object_err(Error::other("direct-memory GET object size is invalid"), vec![bucket, object]))?; diff --git a/crates/heal/src/heal/erasure_healer.rs b/crates/heal/src/heal/erasure_healer.rs index 327f09d15..d4c019c84 100644 --- a/crates/heal/src/heal/erasure_healer.rs +++ b/crates/heal/src/heal/erasure_healer.rs @@ -677,9 +677,7 @@ impl ErasureSetHealer { } *current_object_index = global_obj_idx; - checkpoint_manager - .update_position(bucket_index, *current_object_index) - .await?; + checkpoint_manager.complete_page(bucket_index, *current_object_index).await?; gauge!( "rustfs_heal_page_concurrency_current", "set" => set_disk_id.to_string() diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 47bc6fdae..311174706 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -2371,6 +2371,7 @@ impl HealManager { HealPriority::Low, ); req.source = HealRequestSource::AutoHeal; + req.heal_endpoints = vec![ep.to_string()]; let config = config.read().await; let mut queue = heal_queue.lock().await; let admission = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan"); diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 90f4da057..39c3d2f3f 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -24,9 +24,10 @@ pub mod task; pub mod utils; use storage_api::owner::{ - ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_RUSTFS_META_BUCKET, EcstoreDeleteOptions, EcstoreDiskAPI, - EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, EcstoreErrorType, - EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, ecstore_local_disk_map_read, + ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET, + EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, + EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ObjectIO, ObjectOperations, + ecstore_local_disk_map_read, }; #[cfg(test)] use storage_api::owner::{EcstoreDiskOption, ecstore_new_disk}; @@ -69,6 +70,64 @@ pub async fn clear_unclean_shutdown_markers() { } } +/// Per-disk healing marker path (inside `RUSTFS_META_BUCKET`), mirrored from +/// ecstore so both sides agree on where `DiskInfo.healing` is derived from. +pub(crate) const HEALING_MARKER_PATH: &str = ECSTORE_HEALING_MARKER_PATH; + +/// Write the healing marker on the local disks matching `endpoints` so their +/// `DiskInfo.healing` reports true while the erasure-set heal rebuilds them. +pub(crate) async fn set_healing_markers(endpoints: &[String], set_disk_id: &str) { + apply_healing_markers(endpoints, Some(set_disk_id)).await; +} + +/// Remove the healing markers written by [`set_healing_markers`]. +pub(crate) async fn clear_healing_markers(endpoints: &[String]) { + apply_healing_markers(endpoints, None).await; +} + +async fn apply_healing_markers(endpoints: &[String], set_disk_id: Option<&str>) { + if endpoints.is_empty() { + return; + } + let local_disk_map = local_disk_map_read().await; + for disk in local_disk_map.values().flatten() { + let endpoint = EcstoreDiskAPI::endpoint(disk.as_ref()).to_string(); + if !endpoints.iter().any(|candidate| candidate == &endpoint) { + continue; + } + let result = match set_disk_id { + Some(set_disk_id) => { + EcstoreDiskAPI::write_all( + disk.as_ref(), + RUSTFS_META_BUCKET, + HEALING_MARKER_PATH, + EcstoreDiskBytes::copy_from_slice(set_disk_id.as_bytes()), + ) + .await + } + None => match EcstoreDiskAPI::delete( + disk.as_ref(), + RUSTFS_META_BUCKET, + HEALING_MARKER_PATH, + EcstoreDeleteOptions::default(), + ) + .await + { + Err(DiskError::FileNotFound) => Ok(()), + other => other, + }, + }; + if let Err(err) = result { + tracing::warn!( + endpoint = %endpoint, + action = if set_disk_id.is_some() { "set" } else { "clear" }, + error = ?err, + "failed to update healing marker" + ); + } + } +} + pub(crate) type DiskError = EcstoreDiskError; pub(crate) type DiskResult = EcstoreDiskResult; pub(crate) type DiskStore = EcstoreDiskStore; diff --git a/crates/heal/src/heal/resume.rs b/crates/heal/src/heal/resume.rs index 0d39639c1..3edeefeda 100644 --- a/crates/heal/src/heal/resume.rs +++ b/crates/heal/src/heal/resume.rs @@ -14,9 +14,10 @@ use crate::{Error, Result}; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use std::path::Path; -use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; use tracing::{debug, warn}; use uuid::Uuid; @@ -33,6 +34,39 @@ const RESUME_STATE_FILE: &str = "ahm_resume_state.json"; const RESUME_PROGRESS_FILE: &str = "ahm_progress.json"; const RESUME_CHECKPOINT_FILE: &str = "ahm_checkpoint.json"; +/// Persistence throttle for per-object bookkeeping: flush after this many +/// buffered mutations or once the interval elapses, whichever comes first. +/// Object heal is idempotent, so a crash re-heals at most one throttle window. +const PERSIST_EVERY_MUTATIONS: usize = 1000; +const PERSIST_INTERVAL: Duration = Duration::from_secs(5); + +/// Tracks buffered mutations between persisted snapshots. +#[derive(Debug)] +struct PersistThrottle { + pending: usize, + last_save: Instant, +} + +impl PersistThrottle { + fn new() -> Self { + Self { + pending: 0, + last_save: Instant::now(), + } + } + + /// Record one mutation; returns true when the batch should be flushed. + fn record(&mut self) -> bool { + self.pending += 1; + self.pending >= PERSIST_EVERY_MUTATIONS || self.last_save.elapsed() >= PERSIST_INTERVAL + } + + fn mark_saved(&mut self) { + self.pending = 0; + self.last_save = Instant::now(); + } +} + /// Helper function to convert Path to &str, returning an error if conversion fails fn path_to_str(path: &Path) -> Result<&str> { path.to_str() @@ -168,6 +202,7 @@ impl ResumeState { pub struct ResumeManager { disk: DiskStore, state: Arc>, + throttle: Mutex, } impl ResumeManager { @@ -183,6 +218,7 @@ impl ResumeManager { let manager = Self { disk, state: Arc::new(RwLock::new(state)), + throttle: Mutex::new(PersistThrottle::new()), }; // save initial state @@ -210,6 +246,7 @@ impl ResumeManager { Ok(Self { disk, state: Arc::new(RwLock::new(state)), + throttle: Mutex::new(PersistThrottle::new()), }) } @@ -235,15 +272,31 @@ impl ResumeManager { let mut state = self.state.write().await; state.update_progress(processed, successful, failed, skipped); drop(state); - self.save_state().await + self.save_state_throttled().await } - /// set current item + /// Set current item. Called once per healed object, so persistence is + /// throttled: the in-memory state always updates, but the snapshot is only + /// written every `PERSIST_EVERY_MUTATIONS` calls or `PERSIST_INTERVAL`. pub async fn set_current_item(&self, bucket: Option, object: Option) -> Result<()> { let mut state = self.state.write().await; state.set_current_item(bucket, object); drop(state); - self.save_state().await + let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true); + if !should_save { + return Ok(()); + } + self.save_state_throttled().await + } + + async fn save_state_throttled(&self) -> Result<()> { + let result = self.save_state().await; + if result.is_ok() + && let Ok(mut throttle) = self.throttle.lock() + { + throttle.mark_saved(); + } + result } /// complete bucket @@ -251,7 +304,7 @@ impl ResumeManager { let mut state = self.state.write().await; state.complete_bucket(bucket); drop(state); - self.save_state().await + self.save_state_throttled().await } /// mark task completed @@ -259,7 +312,7 @@ impl ResumeManager { let mut state = self.state.write().await; state.mark_completed(); drop(state); - self.save_state().await + self.save_state_throttled().await } /// set error message @@ -376,12 +429,15 @@ pub struct ResumeCheckpoint { pub current_bucket_index: usize, /// current object index pub current_object_index: usize, - /// processed objects - pub processed_objects: Vec, + /// Objects healed since the last completed page. HashSet: with the + /// previous Vec the per-object `contains` was O(n) and made large-bucket + /// heals O(N²). Only spans the in-flight page — completed pages are + /// covered by `current_object_index`, so `complete_page` prunes the sets. + pub processed_objects: HashSet, /// failed objects - pub failed_objects: Vec, + pub failed_objects: HashSet, /// skipped objects - pub skipped_objects: Vec, + pub skipped_objects: HashSet, } impl ResumeCheckpoint { @@ -391,9 +447,9 @@ impl ResumeCheckpoint { checkpoint_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(), current_bucket_index: 0, current_object_index: 0, - processed_objects: Vec::new(), - failed_objects: Vec::new(), - skipped_objects: Vec::new(), + processed_objects: HashSet::new(), + failed_objects: HashSet::new(), + skipped_objects: HashSet::new(), } } @@ -404,21 +460,25 @@ impl ResumeCheckpoint { } pub fn add_processed_object(&mut self, object: String) { - if !self.processed_objects.contains(&object) { - self.processed_objects.push(object); - } + self.processed_objects.insert(object); } pub fn add_failed_object(&mut self, object: String) { - if !self.failed_objects.contains(&object) { - self.failed_objects.push(object); - } + self.failed_objects.insert(object); } pub fn add_skipped_object(&mut self, object: String) { - if !self.skipped_objects.contains(&object) { - self.skipped_objects.push(object); - } + self.skipped_objects.insert(object); + } + + /// Advance past a fully-processed page: objects below `object_index` are + /// skipped by position on resume, so the per-object sets no longer need + /// their entries and would otherwise grow with the whole bucket. + pub fn complete_page(&mut self, bucket_index: usize, object_index: usize) { + self.update_position(bucket_index, object_index); + self.processed_objects.clear(); + self.skipped_objects.clear(); + self.failed_objects.clear(); } } @@ -426,6 +486,7 @@ impl ResumeCheckpoint { pub struct CheckpointManager { disk: DiskStore, checkpoint: Arc>, + throttle: Mutex, } impl CheckpointManager { @@ -435,6 +496,7 @@ impl CheckpointManager { let manager = Self { disk, checkpoint: Arc::new(RwLock::new(checkpoint)), + throttle: Mutex::new(PersistThrottle::new()), }; // save initial checkpoint @@ -462,6 +524,7 @@ impl CheckpointManager { Ok(Self { disk, checkpoint: Arc::new(RwLock::new(checkpoint)), + throttle: Mutex::new(PersistThrottle::new()), }) } @@ -487,31 +550,59 @@ impl CheckpointManager { let mut checkpoint = self.checkpoint.write().await; checkpoint.update_position(bucket_index, object_index); drop(checkpoint); - self.save_checkpoint().await + self.save_checkpoint_throttled().await } - /// add processed object + /// Advance past a completed page and prune the per-object sets, then persist. + pub async fn complete_page(&self, bucket_index: usize, object_index: usize) -> Result<()> { + let mut checkpoint = self.checkpoint.write().await; + checkpoint.complete_page(bucket_index, object_index); + drop(checkpoint); + self.save_checkpoint_throttled().await + } + + /// Add a processed object. Called once per healed object, so persistence + /// is batched (`PERSIST_EVERY_MUTATIONS` / `PERSIST_INTERVAL`); positions + /// and page boundaries still persist unconditionally. pub async fn add_processed_object(&self, object: String) -> Result<()> { let mut checkpoint = self.checkpoint.write().await; checkpoint.add_processed_object(object); drop(checkpoint); - self.save_checkpoint().await + self.save_checkpoint_if_due().await } - /// add failed object + /// add failed object (batched, see `add_processed_object`) pub async fn add_failed_object(&self, object: String) -> Result<()> { let mut checkpoint = self.checkpoint.write().await; checkpoint.add_failed_object(object); drop(checkpoint); - self.save_checkpoint().await + self.save_checkpoint_if_due().await } - /// add skipped object + /// add skipped object (batched, see `add_processed_object`) pub async fn add_skipped_object(&self, object: String) -> Result<()> { let mut checkpoint = self.checkpoint.write().await; checkpoint.add_skipped_object(object); drop(checkpoint); - self.save_checkpoint().await + self.save_checkpoint_if_due().await + } + + async fn save_checkpoint_if_due(&self) -> Result<()> { + let should_save = self.throttle.lock().map(|mut throttle| throttle.record()).unwrap_or(true); + if !should_save { + return Ok(()); + } + self.save_checkpoint_throttled().await + } + + async fn save_checkpoint_throttled(&self) -> Result<()> { + let result = self.save_checkpoint().await; + if result.is_ok() + && let Ok(mut throttle) = self.throttle.lock() + { + throttle.mark_saved(); + } + result } /// cleanup checkpoint @@ -732,6 +823,55 @@ mod tests { assert!(state.completed_buckets.contains(&"bucket1".to_string())); } + #[test] + fn test_checkpoint_object_sets_dedupe_and_prune() { + let mut checkpoint = ResumeCheckpoint::new("task".to_string()); + checkpoint.add_processed_object("bucket/a".to_string()); + checkpoint.add_processed_object("bucket/a".to_string()); + checkpoint.add_skipped_object("bucket/b".to_string()); + checkpoint.add_failed_object("bucket/c".to_string()); + assert_eq!(checkpoint.processed_objects.len(), 1); + assert!(checkpoint.processed_objects.contains("bucket/a")); + + checkpoint.complete_page(2, 2000); + assert_eq!(checkpoint.current_bucket_index, 2); + assert_eq!(checkpoint.current_object_index, 2000); + assert!(checkpoint.processed_objects.is_empty()); + assert!(checkpoint.skipped_objects.is_empty()); + assert!(checkpoint.failed_objects.is_empty()); + } + + #[test] + fn test_checkpoint_loads_legacy_vec_format() { + // Checkpoints written before the HashSet migration stored the object + // lists as JSON arrays (possibly with duplicates); they must still load. + let legacy = r#"{ + "task_id": "t1", + "checkpoint_time": 1700000000, + "current_bucket_index": 1, + "current_object_index": 42, + "processed_objects": ["a", "b", "a"], + "failed_objects": [], + "skipped_objects": ["c"] + }"#; + let checkpoint: ResumeCheckpoint = serde_json::from_str(legacy).unwrap(); + assert_eq!(checkpoint.current_object_index, 42); + assert_eq!(checkpoint.processed_objects.len(), 2); + assert!(checkpoint.processed_objects.contains("a")); + assert!(checkpoint.skipped_objects.contains("c")); + } + + #[test] + fn test_persist_throttle_batches_until_threshold() { + let mut throttle = PersistThrottle::new(); + for _ in 0..PERSIST_EVERY_MUTATIONS - 1 { + assert!(!throttle.record(), "must not flush below the mutation threshold"); + } + assert!(throttle.record(), "must flush at the mutation threshold"); + throttle.mark_saved(); + assert!(!throttle.record(), "counter must reset after a save"); + } + #[tokio::test] async fn test_resume_utils() { let task_id1 = ResumeUtils::generate_task_id(); diff --git a/crates/heal/src/heal/storage_api.rs b/crates/heal/src/heal/storage_api.rs index ef26703c3..061ca33e0 100644 --- a/crates/heal/src/heal/storage_api.rs +++ b/crates/heal/src/heal/storage_api.rs @@ -17,7 +17,8 @@ pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint; pub(crate) use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult}; pub(crate) use rustfs_ecstore::api::disk::{ BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes, DeleteOptions as EcstoreDeleteOptions, - DiskAPI as EcstoreDiskAPI, DiskStore as EcstoreDiskStore, RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET, + DiskAPI as EcstoreDiskAPI, DiskStore as EcstoreDiskStore, HEALING_MARKER_PATH as ECSTORE_HEALING_MARKER_PATH, + RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET, }; #[cfg(test)] pub(crate) use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, new_disk as ecstore_new_disk}; @@ -30,9 +31,9 @@ pub(crate) mod owner { pub(crate) use super::storage_contracts::{ObjectIO, ObjectOperations}; pub(crate) use super::{ - ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_RUSTFS_META_BUCKET, EcstoreDeleteOptions, - EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, EcstoreEndpoint, - EcstoreErrorType, EcstoreStorageError, EcstoreStore, ecstore_local_disk_map_read, + ECSTORE_BUCKET_META_PREFIX, ECSTORE_DATA_USAGE_CACHE_NAME, ECSTORE_HEALING_MARKER_PATH, ECSTORE_RUSTFS_META_BUCKET, + EcstoreDeleteOptions, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskResult, EcstoreDiskStore, + EcstoreEndpoint, EcstoreErrorType, EcstoreStorageError, EcstoreStore, ecstore_local_disk_map_read, }; #[cfg(test)] diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 4a3e5659b..61dfcf676 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -198,6 +198,11 @@ pub struct HealRequest { pub force_start: bool, /// Number of recoverable retry attempts already scheduled for this request. pub retry_attempts: u32, + /// Endpoints of the disks being rebuilt by an erasure-set heal. Used to + /// write per-disk healing markers so `DiskInfo.healing` reflects reality; + /// empty when the trigger doesn't know the specific disks (admin API, + /// unclean-shutdown verification). + pub heal_endpoints: Vec, /// Created time pub created_at: SystemTime, /// Queue admission time used for scheduler delay metrics @@ -215,6 +220,7 @@ impl HealRequest { source: HealRequestSource::Internal, force_start: false, retry_attempts: 0, + heal_endpoints: Vec::new(), created_at: now, enqueued_at: now, } @@ -267,6 +273,8 @@ pub struct HealTask { pub source: HealRequestSource, /// Number of recoverable retry attempts already scheduled for this task. pub retry_attempts: u32, + /// Endpoints of the disks being rebuilt (see `HealRequest::heal_endpoints`). + pub heal_endpoints: Vec, /// Task status pub status: Arc>, /// Progress tracking @@ -298,6 +306,7 @@ impl HealTask { priority: request.priority, source: request.source, retry_attempts: request.retry_attempts, + heal_endpoints: request.heal_endpoints, status: Arc::new(RwLock::new(HealTaskStatus::Pending)), progress: Arc::new(RwLock::new(HealProgress::new())), result_items: Arc::new(RwLock::new(Vec::new())), @@ -320,6 +329,7 @@ impl HealTask { source: self.source, force_start: false, retry_attempts: self.retry_attempts.saturating_add(1), + heal_endpoints: self.heal_endpoints.clone(), created_at: self.created_at, enqueued_at: SystemTime::now(), } @@ -2038,6 +2048,10 @@ impl HealTask { progress.update_progress(1, 4, 0, 0); } + // The rebuilt disks are formatted now: mark them as healing so + // DiskInfo.healing reflects the rebuild until it completes. + super::set_healing_markers(&self.heal_endpoints, &set_disk_id).await; + // Step 2: Get disk for resume functionality debug!( target: "rustfs::heal::task", @@ -2152,6 +2166,12 @@ impl HealTask { ); let result = erasure_healer.heal_erasure_set(&buckets, &set_disk_id).await; + // Keep the markers on failure: the resume state also persists, and the + // next run of this set heal re-marks and eventually clears them. + if result.is_ok() { + super::clear_healing_markers(&self.heal_endpoints).await; + } + { let mut progress = self.progress.write().await; progress.update_progress(4, 4, 0, 0); diff --git a/rustfs/src/app/context/global.rs b/rustfs/src/app/context/global.rs index da703b664..f3e2e4b4c 100644 --- a/rustfs/src/app/context/global.rs +++ b/rustfs/src/app/context/global.rs @@ -79,6 +79,9 @@ pub struct AppContext { impl AppContext { pub fn new(object_store: Arc, iam: Arc, kms: Arc) -> Self { let object_data_cache = ObjectDataCacheAdapter::from_env_or_disabled(); + // Let ecstore probe this cache inside get_object_reader, after + // metadata resolution but before the erasure data read (backlog#802). + crate::app::object_data_cache::register_object_data_cache_body_hook(Arc::clone(&object_data_cache)); Self { object_store, diff --git a/rustfs/src/app/object_data_cache/hook.rs b/rustfs/src/app/object_data_cache/hook.rs new file mode 100644 index 000000000..ee7b74115 --- /dev/null +++ b/rustfs/src/app/object_data_cache/hook.rs @@ -0,0 +1,143 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! ecstore-facing GET body cache hook. +//! +//! Registered into ecstore so the cache probe runs inside `get_object_reader` +//! right after metadata resolution: probing earlier needs a second metadata +//! fan-out, probing later (after the reader is built) means a hit no longer +//! saves the erasure read/decode. + +use crate::app::object_data_cache::{ + GetObjectBodyCacheLookup, GetObjectBodyCacheRequest, ObjectDataCacheAdapter, build_get_object_body_cache_plan, + lookup_get_object_body_cache_hit, +}; +use crate::app::storage_api::object_usecase::StorageObjectInfo; +use crate::storage::sse::contains_managed_encryption_metadata; +use crate::storage::storage_api::ecstore_bucket::lifecycle::bucket_lifecycle_ops::LifecycleOps as _; +use crate::storage::storage_api::ecstore_object::{GetObjectBodyCacheHook, register_get_object_body_cache_hook}; +use bytes::Bytes; +use rustfs_utils::http::headers::SSEC_ALGORITHM_HEADER; +use std::sync::Arc; + +/// Adapter-backed implementation of ecstore's GET body cache hook. +pub(crate) struct ObjectDataCacheBodyHook { + adapter: Arc, +} + +/// Registers the body-cache hook into ecstore. No-op for a disabled cache so +/// the hot path keeps a single `None` branch when the feature is off. +pub(crate) fn register_object_data_cache_body_hook(adapter: Arc) { + if adapter.is_disabled() { + return; + } + register_get_object_body_cache_hook(Arc::new(ObjectDataCacheBodyHook { adapter })); +} + +fn object_metadata_indicates_encryption(metadata: &std::collections::HashMap) -> bool { + // SSE-C or managed SSE bodies are decrypted on the normal read path, so + // the cached plaintext identity used by the planner would not match what + // ecstore returns here (pre-decryption). Skip them entirely. + metadata.contains_key(SSEC_ALGORITHM_HEADER) || contains_managed_encryption_metadata(metadata) +} + +#[async_trait::async_trait] +impl GetObjectBodyCacheHook for ObjectDataCacheBodyHook { + async fn lookup(&self, bucket: &str, object: &str, info: &StorageObjectInfo) -> Option { + if info.is_remote() || object_metadata_indicates_encryption(&info.user_defined) { + return None; + } + let response_content_length = info.get_actual_size().ok()?; + let request = GetObjectBodyCacheRequest { + bucket, + key: object, + info, + response_content_length, + has_range: false, + part_number: None, + encryption_applied: false, + }; + let plan = build_get_object_body_cache_plan(&self.adapter, request); + match lookup_get_object_body_cache_hit(&self.adapter, &plan).await { + GetObjectBodyCacheLookup::Hit(bytes) => Some(bytes), + GetObjectBodyCacheLookup::Disabled | GetObjectBodyCacheLookup::Skip | GetObjectBodyCacheLookup::Miss => None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_object_data_cache::{ObjectDataCacheConfig, ObjectDataCacheMode}; + + fn hit_only_adapter() -> Arc { + Arc::new( + ObjectDataCacheAdapter::new(ObjectDataCacheConfig { + mode: ObjectDataCacheMode::FillBufferedOnly, + max_bytes: 4 * 1024 * 1024, + ..ObjectDataCacheConfig::default() + }) + .expect("adapter"), + ) + } + + fn plain_info(size: i64) -> StorageObjectInfo { + StorageObjectInfo { + bucket: "b".to_string(), + name: "k".to_string(), + etag: Some("etag-1".to_string()), + size, + actual_size: size, + ..Default::default() + } + } + + #[tokio::test] + async fn hook_lookup_returns_cached_body_after_fill() { + let adapter = hit_only_adapter(); + let info = plain_info(5); + let request = GetObjectBodyCacheRequest { + bucket: "b", + key: "k", + info: &info, + response_content_length: 5, + has_range: false, + part_number: None, + encryption_applied: false, + }; + let plan = build_get_object_body_cache_plan(&adapter, request); + let body = Bytes::from_static(b"hello"); + let _ = crate::app::object_data_cache::fill_get_object_body_cache_from_buffered_body(&adapter, &plan, &body).await; + + let hook = ObjectDataCacheBodyHook { + adapter: Arc::clone(&adapter), + }; + let hit = hook.lookup("b", "k", &info).await.expect("cache hit"); + assert_eq!(hit, body); + } + + #[tokio::test] + async fn hook_lookup_skips_encrypted_objects() { + let adapter = hit_only_adapter(); + let mut info = plain_info(5); + info.user_defined = std::sync::Arc::new( + [(SSEC_ALGORITHM_HEADER.to_string(), "AES256".to_string())] + .into_iter() + .collect(), + ); + + let hook = ObjectDataCacheBodyHook { adapter }; + assert!(hook.lookup("b", "k", &info).await.is_none()); + } +} diff --git a/rustfs/src/app/object_data_cache/mod.rs b/rustfs/src/app/object_data_cache/mod.rs index 922d22af1..006d2f4ad 100644 --- a/rustfs/src/app/object_data_cache/mod.rs +++ b/rustfs/src/app/object_data_cache/mod.rs @@ -16,6 +16,7 @@ mod adapter; mod body; +mod hook; mod invalidation; mod planner; @@ -24,6 +25,7 @@ pub(crate) use body::{ GetObjectBodyCacheLookup, fill_get_object_body_cache_from_buffered_body, fill_get_object_body_cache_from_materialized_body, lookup_get_object_body_cache_hit, }; +pub(crate) use hook::register_object_data_cache_body_hook; pub(crate) use invalidation::{ invalidate_object_data_cache_after_complete_multipart_success, invalidate_object_data_cache_after_copy_success, invalidate_object_data_cache_after_delete_success, invalidate_object_data_cache_after_put_success, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 1e8ba41ee..436d0d072 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -330,7 +330,6 @@ struct GetObjectRequestContext { struct GetObjectReadSetup { info: ObjectInfo, - event_info: ObjectInfo, final_stream: DynReader, buffered_body: Option, rs: Option, @@ -361,7 +360,7 @@ struct GetObjectStrategyContext { struct GetObjectOutputContext { output: GetObjectOutput, - event_info: ObjectInfo, + event_info: Option, response_content_length: i64, optimal_buffer_size: usize, } @@ -522,7 +521,7 @@ pin_project! { struct DiskReadPermitReader { #[pin] inner: R, - _disk_permit: OwnedSemaphorePermit, + disk_permit: Option, } } @@ -530,7 +529,7 @@ impl DiskReadPermitReader { fn new(inner: R, disk_permit: OwnedSemaphorePermit) -> Self { Self { inner, - _disk_permit: disk_permit, + disk_permit: Some(disk_permit), } } } @@ -540,7 +539,16 @@ where R: AsyncRead, { fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - self.project().inner.poll_read(cx, buf) + let this = self.project(); + let had_capacity = buf.remaining() > 0; + let filled_before = buf.filled().len(); + let poll = this.inner.poll_read(cx, buf); + // EOF: no more disk reads can happen through this stream, so release + // the permit instead of holding it until the client drops the body. + if had_capacity && matches!(poll, Poll::Ready(Ok(()))) && buf.filled().len() == filled_before { + this.disk_permit.take(); + } + poll } } @@ -2208,6 +2216,18 @@ impl DefaultObjectUsecase { }) } + /// How long a GET waits for a disk read permit before degrading to a + /// permit-less read. Cached: consulted per GET. Zero disables the bound. + fn disk_permit_wait_timeout() -> Duration { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_OBJECT_DISK_PERMIT_WAIT_TIMEOUT, + rustfs_config::DEFAULT_OBJECT_DISK_PERMIT_WAIT_TIMEOUT, + )) + }) + } + async fn acquire_get_object_io_planning( manager: &ConcurrencyManager, wrapper: &RequestTimeoutWrapper, @@ -2216,10 +2236,32 @@ impl DefaultObjectUsecase { key: &str, ) -> S3Result { let permit_wait_start = std::time::Instant::now(); - let disk_permit = manager - .acquire_owned_disk_read_permit() - .await - .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?; + let permit_wait_timeout = Self::disk_permit_wait_timeout(); + // Permits are held for the whole body transfer, so slow clients can + // pin all of them while disks are idle. Bound the wait and degrade to + // a permit-less read instead of stalling into the request timeout. + let disk_permit = if permit_wait_timeout.is_zero() { + Some( + manager + .acquire_owned_disk_read_permit() + .await + .map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?, + ) + } else { + match tokio::time::timeout(permit_wait_timeout, manager.acquire_owned_disk_read_permit()).await { + Ok(permit) => Some(permit.map_err(|_| s3_error!(InternalError, "disk read semaphore closed"))?), + Err(_) => { + metrics::counter!("rustfs.get_object.disk_permit.bypass.total").increment(1); + warn!( + bucket = %bucket, + key = %key, + wait_ms = permit_wait_start.elapsed().as_millis() as u64, + "GetObject proceeding without disk read permit after bounded wait" + ); + None + } + } + }; let permit_wait_duration = permit_wait_start.elapsed(); Self::ensure_get_object_not_timed_out( @@ -2253,7 +2295,7 @@ impl DefaultObjectUsecase { Self::ensure_get_object_not_timed_out(wrapper, timeout_config, bucket, key, GetObjectTimeoutStage::BeforeRead)?; Ok(GetObjectIoPlanning { - disk_permit: Some(disk_permit), + disk_permit, permit_wait_duration, queue_status, queue_utilization, @@ -2261,14 +2303,12 @@ impl DefaultObjectUsecase { } async fn prepare_get_object_request_context(req: &S3Request) -> S3Result { - let GetObjectInput { - bucket, - key, - version_id, - part_number, - range, - .. - } = req.input.clone(); + // Clone only the fields this path needs instead of the whole input. + let bucket = req.input.bucket.clone(); + let key = req.input.key.clone(); + let version_id = req.input.version_id.clone(); + let part_number = req.input.part_number; + let range = req.input.range; validate_object_key(&key, "GET")?; @@ -2420,7 +2460,6 @@ impl DefaultObjectUsecase { ); } - let event_info = info.clone(); let content_type = if let Some(content_type) = &info.content_type { match ContentType::from_str(content_type) { Ok(res) => Some(res), @@ -2510,7 +2549,6 @@ impl DefaultObjectUsecase { Ok(GetObjectReadSetup { info, - event_info, final_stream, buffered_body, rs, @@ -2595,8 +2633,6 @@ impl DefaultObjectUsecase { request_size = response_content_length, "I/O priority assigned (based on actual request size)" ); - - rustfs_io_metrics::record_io_priority_assignment(io_priority.as_str()); } rustfs_io_metrics::record_get_object_io_state( @@ -3609,11 +3645,15 @@ impl DefaultObjectUsecase { bucket: &str, method: &hyper::Method, headers: &HeaderMap, - event_info: ObjectInfo, + event_info: Option, version_id_for_event: String, output: GetObjectOutput, ) -> S3Result> { - let helper = helper.object(event_info).version_id(version_id_for_event); + let helper = match event_info { + Some(event_info) => helper.object(event_info), + None => helper, + }; + let helper = helper.version_id(version_id_for_event); let response = wrap_response_with_cors(bucket, method, headers, output).await; let result = Ok(response); let _ = helper.complete(&result); @@ -3627,7 +3667,7 @@ impl DefaultObjectUsecase { bucket: &str, key: &str, info: ObjectInfo, - event_info: ObjectInfo, + event_info: Option, final_stream: DynReader, buffered_body: Option, rs: Option, @@ -3831,7 +3871,6 @@ impl DefaultObjectUsecase { let GetObjectReadSetup { info, - event_info, final_stream, buffered_body, rs, @@ -3852,6 +3891,10 @@ impl DefaultObjectUsecase { final_stream }; + // Clone ObjectInfo for event notification only when an event will + // actually be built — the clone is expensive for multipart objects. + let event_info = helper.wants_object_info().then(|| info.clone()); + let output_build_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); let output_context = self .build_get_object_output_context( @@ -5225,7 +5268,9 @@ impl DefaultObjectUsecase { // Compute x-amz-expiration header from lifecycle prediction (before info is partially moved) let expiration_header = resolve_put_object_expiration(&bucket, &info).await; - let event_info = info.clone(); + // Clone ObjectInfo for event notification only when an event will + // actually be built — the clone is expensive for multipart objects. + let event_info = helper.wants_object_info().then(|| info.clone()); let content_type = { if let Some(content_type) = &info.content_type { match ContentType::from_str(content_type) { @@ -5344,7 +5389,10 @@ impl DefaultObjectUsecase { }; let version_id = req.input.version_id.clone().unwrap_or_default(); - helper = helper.object(event_info).version_id(version_id); + if let Some(event_info) = event_info { + helper = helper.object(event_info); + } + helper = helper.version_id(version_id); // NOTE ON CORS: // Bucket-level CORS headers are intentionally applied only for object retrieval @@ -7401,6 +7449,26 @@ mod tests { assert_eq!(body, b"hello"); } + #[tokio::test] + async fn disk_read_permit_reader_releases_permit_at_eof() { + use tokio::io::AsyncReadExt; + + let semaphore = Arc::new(tokio::sync::Semaphore::new(1)); + let permit = semaphore.clone().acquire_owned().await.expect("acquire permit"); + assert_eq!(semaphore.available_permits(), 0); + + let mut reader = DiskReadPermitReader::new(std::io::Cursor::new(b"hello".to_vec()), permit); + let mut body = Vec::new(); + reader.read_to_end(&mut body).await.expect("read body"); + assert_eq!(body, b"hello"); + + // The reader is still alive (client hasn't dropped the body), but EOF + // was observed, so the permit must already be back in the semaphore. + assert_eq!(semaphore.available_permits(), 1); + drop(reader); + assert_eq!(semaphore.available_permits(), 1); + } + #[tokio::test] async fn pooled_buffer_reader_keeps_buffer_alive_until_consumed() { use tokio::io::AsyncReadExt; @@ -7783,7 +7851,7 @@ mod tests { "test-bucket", "path/raw", info.clone(), - info, + Some(info), wrap_reader(tokio::io::empty()), None, None, diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index c960d1c68..4918c6570 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -237,6 +237,13 @@ impl OperationHelper { })) } + /// True when a pending event notification still needs the final ObjectInfo. + /// Callers can use this to avoid cloning ObjectInfo when the event chain is + /// disabled or suppressed. + pub fn wants_object_info(&self) -> bool { + matches!(self, Self::Enabled(state) if state.event_builder.is_some()) + } + /// Sets the ObjectInfo for event notification. pub fn object(mut self, object_info: ObjectInfo) -> Self { if let Self::Enabled(state) = &mut self diff --git a/rustfs/src/storage/options.rs b/rustfs/src/storage/options.rs index adc84d550..eac029fd6 100644 --- a/rustfs/src/storage/options.rs +++ b/rustfs/src/storage/options.rs @@ -45,7 +45,44 @@ use crate::auth::AuthType; use crate::auth::get_query_param; use crate::auth::get_request_auth_type_with_query; use crate::auth::is_request_presigned_signature_v4_with_query; +use crate::storage::storage_api::ecstore_bucket::versioning::VersioningApi as _; use crate::storage::storage_api::options_consumer::StorageObjectOptions as ObjectOptions; +use s3s::dto::VersioningConfiguration; + +/// Fetch the bucket's versioning configuration once so callers can derive +/// enabled/suspended state without repeated metadata-sys lookups per request. +async fn bucket_versioning_config(bucket: &str) -> VersioningConfiguration { + match BucketVersioningSys::get(bucket).await { + Ok(cfg) => cfg, + Err(err) => { + tracing::warn!("{:?}", err); + VersioningConfiguration::default() + } + } +} + +/// Whether GET should skip per-shard bitrot verification. Read once: the env +/// flag is consulted on every GET and `std::env::var` takes a process-global +/// lock. In tests the env is read directly so `temp_env` overrides apply. +fn get_skip_verify_bitrot() -> bool { + #[cfg(test)] + { + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY, + rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY, + ) + } + #[cfg(not(test))] + { + static CACHED: std::sync::OnceLock = std::sync::OnceLock::new(); + *CACHED.get_or_init(|| { + rustfs_utils::get_env_bool( + rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY, + rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY, + ) + }) + } +} /// Creates options for deleting an object in a bucket. pub async fn del_opts( @@ -55,8 +92,9 @@ pub async fn del_opts( headers: &HeaderMap, metadata: HashMap, ) -> Result { - let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await; - let version_suspended = BucketVersioningSys::suspended(bucket).await; + let versioning_cfg = bucket_versioning_config(bucket).await; + let versioned = versioning_cfg.prefix_enabled(object); + let version_suspended = versioning_cfg.suspended(); let vid = if vid.is_none() { get_header(headers, SUFFIX_SOURCE_VERSION_ID).map(|s| s.into_owned()) @@ -120,8 +158,9 @@ pub async fn get_opts( part_num: Option, headers: &HeaderMap, ) -> Result { - let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await; - let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await; + let versioning_cfg = bucket_versioning_config(bucket).await; + let versioned = versioning_cfg.prefix_enabled(object); + let version_suspended = versioning_cfg.prefix_suspended(object); let vid = vid.map(|v| v.as_str().trim().to_owned()); @@ -159,10 +198,7 @@ pub async fn get_opts( // Optionally skip per-shard bitrot hash verification on reads to save CPU. // Background scanner still performs full integrity checks asynchronously. - opts.skip_verify_bitrot = rustfs_utils::get_env_bool( - rustfs_config::ENV_OBJECT_GET_SKIP_BITROT_VERIFY, - rustfs_config::DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY, - ); + opts.skip_verify_bitrot = get_skip_verify_bitrot(); fill_conditional_writes_opts_from_header(headers, &mut opts)?; @@ -213,8 +249,9 @@ pub async fn put_opts( headers: &HeaderMap, metadata: HashMap, ) -> Result { - let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await; - let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await; + let versioning_cfg = bucket_versioning_config(bucket).await; + let versioned = versioning_cfg.prefix_enabled(object); + let version_suspended = versioning_cfg.prefix_suspended(object); let vid = if vid.is_none() { get_header(headers, SUFFIX_SOURCE_VERSION_ID).map(|s| s.into_owned()) diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index b2df5ae13..498335b86 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -2208,7 +2208,7 @@ pub fn mark_encrypted_multipart_metadata(metadata: &mut HashMap) metadata.insert(MINIO_INTERNAL_ENCRYPTION_MULTIPART_HEADER.to_string(), String::new()); } -fn contains_managed_encryption_metadata(metadata: &HashMap) -> bool { +pub(crate) fn contains_managed_encryption_metadata(metadata: &HashMap) -> bool { metadata.contains_key(INTERNAL_ENCRYPTION_KEY_HEADER) || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_S3_SEALED_KEY_HEADER) || metadata.contains_key(MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER) diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 376c584a9..94934ce4a 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -431,6 +431,10 @@ pub(crate) mod ecstore_rpc { }; } +pub(crate) mod ecstore_object { + pub(crate) use rustfs_ecstore::api::object::{GetObjectBodyCacheHook, register_get_object_body_cache_hook}; +} + pub(crate) mod ecstore_set_disk { pub(crate) use rustfs_ecstore::api::set_disk::{DEFAULT_READ_BUFFER_SIZE, get_lock_acquire_timeout, is_valid_storage_class}; }