From cfee2ffabacc51adf996dbf9f5d4caf51bb551eb Mon Sep 17 00:00:00 2001 From: Ramakrishna Chilaka <49393831+RamakrishnaChilaka@users.noreply.github.com> Date: Fri, 23 Oct 2026 20:02:48 +0530 Subject: [PATCH] perf(ecstore): bound xl.meta read workers and enable HEAD early-stop (#8846) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a --- crates/ecstore/src/disk/local.rs | 413 ++++++++++++++++-- .../src/set_disk/core/metadata_read.rs | 37 +- crates/ecstore/src/set_disk/mod.rs | 154 ++++++- crates/ecstore/src/set_disk/ops/object.rs | 48 +- crates/ecstore/src/set_disk/read.rs | 6 +- 5 files changed, 597 insertions(+), 61 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1df81adab..d3da31056 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -273,6 +273,169 @@ fn read_all_data_std( Ok((bytes, modtime)) } +fn read_metadata_only_std(opened: std::io::Result) -> Result<(Vec, Option)> { + let mut f = opened.map_err(to_file_error)?; + + let meta = f.metadata().map_err(to_file_error)?; + + if meta.is_dir() { + return Err(Error::FileNotFound); + } + + let size = usize::try_from(meta.len()).map_err(Error::other)?; + + let data = read_xl_meta_no_data_sync(&mut f, size)?; + + let modtime = match meta.modified() { + Ok(md) => Some(OffsetDateTime::from(md)), + Err(_) => None, + }; + + Ok((data, modtime)) +} + +const MAX_METADATA_READ_WORKERS_PER_DISK: usize = 2; + +type ReadAllDataResult = core::result::Result<(Vec, Option), ReadAllError>; +type ReadMetadataOnlyResult = Result<(Vec, Option)>; + +enum MetadataReadOp { + ReadAll { + tx: tokio::sync::oneshot::Sender, + }, + ReadMetadataOnly { + tx: tokio::sync::oneshot::Sender, + }, + #[cfg(test)] + TestCustom { + is_closed: Box bool + Send>, + run: Box, + }, +} + +struct MetadataReadJob { + path: PathBuf, + io_permit: Option, + op: MetadataReadOp, +} + +impl MetadataReadJob { + #[inline] + fn is_cancelled(&self) -> bool { + match &self.op { + MetadataReadOp::ReadAll { tx } => tx.is_closed(), + MetadataReadOp::ReadMetadataOnly { tx } => tx.is_closed(), + #[cfg(test)] + MetadataReadOp::TestCustom { is_closed, .. } => is_closed(), + } + } + + #[inline] + fn execute(self, root: &os::PublicationRoot) { + if self.is_cancelled() { + return; + } + let _io_permit = self.io_permit; + match self.op { + MetadataReadOp::ReadAll { tx } => { + let _ = tx.send(read_all_data_std(root.open_readonly(&self.path))); + } + MetadataReadOp::ReadMetadataOnly { tx } => { + let _ = tx.send(read_metadata_only_std(root.open_readonly(&self.path))); + } + #[cfg(test)] + MetadataReadOp::TestCustom { run, .. } => run(), + } + } +} + +#[derive(Default)] +struct MetadataReadQueueState { + jobs: std::collections::VecDeque, + active_workers: usize, +} + +struct MetadataReadQueue { + root: os::PublicationRoot, + state: ParkingLotMutex, +} + +struct MetadataReadWorkerGuard { + queue: Arc, + disarmed: bool, +} + +impl Drop for MetadataReadWorkerGuard { + fn drop(&mut self) { + if self.disarmed { + return; + } + let should_respawn = { + let mut state = self.queue.state.lock(); + state.active_workers = state.active_workers.saturating_sub(1); + if !state.jobs.is_empty() && state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK { + state.active_workers += 1; + true + } else { + false + } + }; + if should_respawn { + MetadataReadQueue::spawn_worker(Arc::clone(&self.queue)); + } + } +} + +impl MetadataReadQueue { + fn new(root: os::PublicationRoot) -> Arc { + Arc::new(Self { + root, + state: ParkingLotMutex::new(MetadataReadQueueState::default()), + }) + } + + fn enqueue(self: &Arc, job: MetadataReadJob) { + let should_spawn = { + let mut state = self.state.lock(); + state.jobs.push_back(job); + if state.active_workers < MAX_METADATA_READ_WORKERS_PER_DISK { + state.active_workers += 1; + true + } else { + false + } + }; + if should_spawn { + Self::spawn_worker(Arc::clone(self)); + } + } + + fn spawn_worker(queue: Arc) { + drop(tokio::task::spawn_blocking(move || { + let mut guard = MetadataReadWorkerGuard { + queue: Arc::clone(&queue), + disarmed: false, + }; + loop { + let Some(job) = ({ + let mut state = queue.state.lock(); + match state.jobs.pop_front() { + Some(job) => Some(job), + None => { + state.active_workers = state.active_workers.saturating_sub(1); + guard.disarmed = true; + None + } + } + }) else { + return; + }; + job.execute(&queue.root); + } + })); + } +} + /// Reads the `xl.meta` header/metadata envelope and 5-byte CRC for directory /// walks (matching `read_xl_meta_no_data_sync`) without reading trailing /// `InlineData` object bodies. If the header or metadata envelope is corrupt or @@ -5586,6 +5749,7 @@ pub struct LocalDisk { exit_signal: Option>, io_backend: Arc, file_sync_permits: Arc, + metadata_read_queue: Arc, snapshot_leases: Arc>, } @@ -6041,6 +6205,7 @@ impl LocalDisk { // TODO(backlog): add O_DIRECT I/O support for performance-critical paths // TODO(backlog): populate DiskInfo in constructor + let metadata_read_queue = MetadataReadQueue::new(publication_root.clone()); let mut disk = Self { root: root.clone(), publication_root, @@ -6071,6 +6236,7 @@ impl LocalDisk { exit_signal: None, io_backend: build_local_io_backend(io_root.clone()).await, file_sync_permits: os::disk_file_sync_limiter(&root), + metadata_read_queue, snapshot_leases: Arc::new(ParkingLotMutex::new(SnapshotLeaseRegistry::default())), }; let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| { @@ -6563,6 +6729,26 @@ impl LocalDisk { self.io_get_object_path(bucket, key) } + /// Only for operations that open through PublicationRoot or a commit guard: + /// containment and symlink checks happen with the actual open, off the runtime. + fn io_get_bucket_and_object_open_path(&self, bucket: &str, key: &str) -> Result<(PathBuf, PathBuf)> { + #[cfg(target_os = "linux")] + { + let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key); + let normalized = check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?; + if matches!(normalized, std::borrow::Cow::Owned(_)) { + reject_local_disk_symlink_components_at(self.io_root(), &self.mount_lease, &normalized)?; + } + Ok((bucket_path, path)) + } + #[cfg(not(target_os = "linux"))] + { + let path = self.io_get_object_path(bucket, key)?; + let bucket_path = self.io_get_bucket_path(bucket)?; + Ok((bucket_path, path)) + } + } + fn io_get_bucket_path(&self, bucket: &str) -> Result { self.local_disk_bucket_path(self.io_root(), bucket) } @@ -7984,32 +8170,13 @@ impl LocalDisk { // - metadata failure -> to_file_error // - parse failure -> propagated verbatim from read_xl_meta_no_data_sync (`?`) let path = file_path.as_ref().to_path_buf(); - let root = self.publication_root.clone(); - let (data, modtime) = - spawn_blocking_with_disk_io_permit(io_permit, move || -> Result<(Vec, Option)> { - // Read-only open, equivalent to O_RDONLY (get_readonly_options only sets read(true)). - let mut f = root.open_readonly(&path).map_err(to_file_error)?; - - let meta = f.metadata().map_err(to_file_error)?; - - if meta.is_dir() { - // fix use io::Error - return Err(Error::FileNotFound); - } - - let size = meta.len() as usize; - - let data = read_xl_meta_no_data_sync(&mut f, size)?; - - let modtime = match meta.modified() { - Ok(md) => Some(OffsetDateTime::from(md)), - Err(_) => None, - }; - - Ok((data, modtime)) - }) - .await - .map_err(DiskError::from)??; + let (tx, rx) = tokio::sync::oneshot::channel(); + self.metadata_read_queue.enqueue(MetadataReadJob { + path, + io_permit, + op: MetadataReadOp::ReadMetadataOnly { tx }, + }); + let (data, modtime) = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))??; Ok((data, modtime)) } @@ -8180,10 +8347,13 @@ impl LocalDisk { // gating the volume fallback on the open error alone is equivalent to // the original code, where the fallback lived solely in the open match arm. let path = file_path.as_ref().to_path_buf(); - let root = self.publication_root.clone(); - let res = spawn_blocking_with_disk_io_permit(io_permit, move || read_all_data_std(root.open_readonly(&path))) - .await - .map_err(DiskError::from)?; + let (tx, rx) = tokio::sync::oneshot::channel(); + self.metadata_read_queue.enqueue(MetadataReadJob { + path, + io_permit, + op: MetadataReadOp::ReadAll { tx }, + }); + let res = rx.await.map_err(|_| DiskError::other("metadata read worker cancelled"))?; let (bytes, modtime) = self.resolve_read_all_result(volume, volume_dir.as_ref(), res).await?; @@ -10127,19 +10297,23 @@ fn check_local_disk_valid_path(root: &Path, path: impl AsRef) -> Result<() #[cfg(target_os = "linux")] fn check_local_disk_valid_object_path_at(root: &Path, root_fd: &std::fs::File, bucket_path: &Path, path: &Path) -> Result<()> { - check_local_disk_object_path_components(root, bucket_path, path)?; - reject_local_disk_symlink_components_at(root, root_fd, &normalize_path_components(path)) + let normalized_path = check_local_disk_object_path_components(root, bucket_path, path)?; + reject_local_disk_symlink_components_at(root, root_fd, &normalized_path) } #[cfg(target_os = "linux")] -fn check_local_disk_object_path_components(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> { - let bucket_path = normalize_path_components(bucket_path); - let path = normalize_path_components(path); +fn check_local_disk_object_path_components<'a>( + root: &Path, + bucket_path: &Path, + path: &'a Path, +) -> Result> { + let bucket_path = normalize_path_cow(bucket_path); + let path = normalize_path_cow(path); if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) { return Err(DiskError::InvalidPath); } - Ok(()) + Ok(path) } #[cfg(target_os = "linux")] @@ -10226,6 +10400,18 @@ fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()> Ok(()) } +#[cfg(target_os = "linux")] +#[inline] +fn normalize_path_cow(path: &Path) -> std::borrow::Cow<'_, Path> { + if !path + .components() + .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir)) + { + return std::borrow::Cow::Borrowed(path); + } + std::borrow::Cow::Owned(normalize_path_components(path)) +} + // Lightweight path normalization without filesystem calls fn normalize_path_components(path: impl AsRef) -> PathBuf { let path = path.as_ref(); @@ -12428,8 +12614,7 @@ impl DiskAPI for LocalDisk { } let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let file_path = self.io_get_object_path(volume, path)?; - let volume_dir = self.io_get_bucket_path(volume)?; + let (volume_dir, file_path) = self.io_get_bucket_and_object_open_path(volume, path)?; record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start); let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled); @@ -12439,7 +12624,7 @@ impl DiskAPI for LocalDisk { let read_data = opts.read_data; let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled); - let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await; + let raw_read_result = self.read_raw(volume, &volume_dir, &file_path, read_data).await; record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start); let (data, _) = raw_read_result.map_err(|e| { if e == DiskError::FileNotFound && !version_id.is_empty() { @@ -25523,6 +25708,156 @@ mod test { assert!(matches!(disk.get_object_path_for_io("bucket", "object"), Err(DiskError::InvalidPath))); } + #[cfg(unix)] + #[tokio::test] + async fn read_version_rejects_symlink_prefix_and_leaf() { + use std::os::unix::fs::symlink; + + let root_dir = tempfile::tempdir().expect("temp dir should be created"); + let outside_dir = tempfile::tempdir().expect("outside temp dir should be created"); + let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + disk.make_volume("bucket").await.expect("bucket should be created"); + + let valid_meta = test_meta(test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline")))); + fs::create_dir_all(outside_dir.path().join("object")) + .await + .expect("outside object dir should be created"); + fs::write(outside_dir.path().join("object").join(STORAGE_FORMAT_FILE), &valid_meta) + .await + .expect("outside prefix xl.meta should be written"); + fs::write(outside_dir.path().join(STORAGE_FORMAT_FILE), &valid_meta) + .await + .expect("outside leaf xl.meta should be written"); + + symlink(outside_dir.path(), root_dir.path().join("bucket/symlink-prefix")).expect("prefix symlink should be created"); + symlink(outside_dir.path(), root_dir.path().join("bucket/symlink-leaf")).expect("leaf symlink should be created"); + + for read_data in [false, true] { + let opts = ReadOptions { + read_data, + ..Default::default() + }; + assert!(matches!( + disk.read_version("", "bucket", "symlink-prefix/object", "", &opts).await, + Err(DiskError::InvalidPath) + )); + assert!(matches!( + disk.read_version("", "bucket", "symlink-leaf", "", &opts).await, + Err(DiskError::InvalidPath) + )); + } + } + + #[tokio::test] + async fn metadata_read_queue_bounds_concurrency_skips_cancelled_and_recovers_from_panic() { + let dir = tempfile::tempdir().expect("tempdir should be created"); + let root = os::PublicationRoot::new(dir.path()).expect("publication root should be created"); + let queue = MetadataReadQueue::new(root); + + // 1. Concurrency bound + cancelled job skip: + let active = Arc::new(AtomicUsize::new(0)); + let max_active = Arc::new(AtomicUsize::new(0)); + let executed_cancelled = Arc::new(AtomicBool::new(false)); + let (entered_tx, mut entered_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let release_rx = Arc::new(ParkingLotMutex::new(release_rx)); + + for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK { + let active = Arc::clone(&active); + let max_active = Arc::clone(&max_active); + let entered_tx = entered_tx.clone(); + let release_rx = Arc::clone(&release_rx); + queue.enqueue(MetadataReadJob { + path: PathBuf::new(), + io_permit: None, + op: MetadataReadOp::TestCustom { + is_closed: Box::new(|| false), + run: Box::new(move || { + let cur = active.fetch_add(1, Ordering::AcqRel) + 1; + max_active.fetch_max(cur, Ordering::AcqRel); + let _ = entered_tx.send(()); + let _ = release_rx.lock().recv(); + active.fetch_sub(1, Ordering::AcqRel); + }), + }, + }); + } + + for _ in 0..MAX_METADATA_READ_WORKERS_PER_DISK { + tokio::time::timeout(Duration::from_secs(5), entered_rx.recv()) + .await + .expect("initial workers should enter") + .expect("channel should stay open"); + } + assert_eq!(queue.state.lock().active_workers, MAX_METADATA_READ_WORKERS_PER_DISK); + + // Enqueue a cancelled production ReadMetadataOnly job and a cancelled TestCustom job while workers are busy. + let (cancelled_tx, cancelled_rx) = tokio::sync::oneshot::channel(); + drop(cancelled_rx); + queue.enqueue(MetadataReadJob { + path: dir.path().join("cancelled-never-opened"), + io_permit: None, + op: MetadataReadOp::ReadMetadataOnly { tx: cancelled_tx }, + }); + let executed_cancelled_clone = Arc::clone(&executed_cancelled); + queue.enqueue(MetadataReadJob { + path: PathBuf::new(), + io_permit: None, + op: MetadataReadOp::TestCustom { + is_closed: Box::new(|| true), + run: Box::new(move || { + executed_cancelled_clone.store(true, Ordering::Release); + }), + }, + }); + + // Enqueue a panicking job followed by a normal completion job to verify panic respawn. + queue.enqueue(MetadataReadJob { + path: PathBuf::new(), + io_permit: None, + op: MetadataReadOp::TestCustom { + is_closed: Box::new(|| false), + run: Box::new(|| panic!("synthetic worker panic")), + }, + }); + let (done_tx, done_rx) = tokio::sync::oneshot::channel(); + let active_trailing = Arc::clone(&active); + let max_active_trailing = Arc::clone(&max_active); + queue.enqueue(MetadataReadJob { + path: PathBuf::new(), + io_permit: None, + op: MetadataReadOp::TestCustom { + is_closed: Box::new(|| false), + run: Box::new(move || { + let cur = active_trailing.fetch_add(1, Ordering::AcqRel) + 1; + max_active_trailing.fetch_max(cur, Ordering::AcqRel); + active_trailing.fetch_sub(1, Ordering::AcqRel); + let _ = done_tx.send(()); + }), + }, + }); + + assert_eq!( + queue.state.lock().active_workers, + MAX_METADATA_READ_WORKERS_PER_DISK, + "extra queued jobs must not exceed per-disk worker cap" + ); + + // Release only one blocked worker so the panicking job must trigger a respawn + // to execute the trailing completion job while the other worker stays blocked. + release_tx.send(()).expect("release one blocked worker"); + tokio::time::timeout(Duration::from_secs(5), done_rx) + .await + .expect("trailing job should finish after worker panic recovery") + .expect("trailing job sender should complete"); + for _ in 1..MAX_METADATA_READ_WORKERS_PER_DISK { + release_tx.send(()).expect("release remaining blocked worker"); + } + assert!(!executed_cancelled.load(Ordering::Acquire), "cancelled job must be skipped"); + assert!(max_active.load(Ordering::Acquire) <= MAX_METADATA_READ_WORKERS_PER_DISK); + } + #[tokio::test] async fn get_object_path_rejects_key_traversal_out_of_bucket() { let root_dir = tempfile::tempdir().expect("temp dir should be created"); diff --git a/crates/ecstore/src/set_disk/core/metadata_read.rs b/crates/ecstore/src/set_disk/core/metadata_read.rs index e3d318589..5aa1e60ae 100644 --- a/crates/ecstore/src/set_disk/core/metadata_read.rs +++ b/crates/ecstore/src/set_disk/core/metadata_read.rs @@ -67,6 +67,7 @@ pub(in crate::set_disk) struct MetadataReadResult { pub(in crate::set_disk) slots: Vec, pub(in crate::set_disk) diagnostics: MetadataFanoutDiagnostics, pub(in crate::set_disk) condition_stopped: bool, + pub(in crate::set_disk) verified_inline_body: Option, } impl MetadataReadResult { @@ -634,7 +635,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason( parts_metadata: &[FileInfo], disks: &[Option], ) -> Option<&'static str> { - inline_body_miss_reason_from_observations( + verify_inline_body_from_observations( bucket, object, candidate, @@ -645,17 +646,18 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason( disks, ) .await + .err() } -async fn inline_body_miss_reason_from_observations<'a>( +async fn verify_inline_body_from_observations<'a>( bucket: &str, object: &str, candidate: &FileInfo, observations: impl IntoIterator)>, disks: &[Option], -) -> Option<&'static str> { +) -> Result { if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) { - return Some(reason); + return Err(reason); } let Ok(erasure) = coding::Erasure::try_new_with_options( @@ -664,21 +666,21 @@ async fn inline_body_miss_reason_from_observations<'a>( candidate.erasure.block_size, candidate.uses_legacy_checksum, ) else { - return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY); + return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY); }; let data_files = match collect_inline_data_shard_fileinfos_from_observations(observations, candidate, erasure.data_shards, |index| { disks.get(index).is_some_and(Option::is_some) }) { Ok(data_files) => data_files, - Err(reason) => return Some(reason), + Err(reason) => return Err(reason), }; let Some(part) = candidate.parts.first() else { - return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE); + return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE); }; let Ok(object_size) = usize::try_from(candidate.size) else { - return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE); + return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE); }; let checksum_info = candidate.erasure.get_checksum_info(part.number); let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S { @@ -698,14 +700,14 @@ async fn inline_body_miss_reason_from_observations<'a>( let Ok(mut readers) = build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await else { - return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY); + return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY); }; #[cfg(test)] crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0); match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await { - Some(body) if body.len() == object_size => None, - _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY), + Some(body) if body.len() == object_size => Ok(body), + _ => Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY), } } @@ -1148,6 +1150,7 @@ impl SetDisks { slots, diagnostics, condition_stopped: false, + verified_inline_body: None, }) } @@ -1368,6 +1371,7 @@ impl SetDisks { ), ) }); + let mut verified_inline_body = None; let should_return_early = if condition_stopped { true } else if force_full_wait { @@ -1377,7 +1381,7 @@ impl SetDisks { Some(_candidate) if non_inline_candidate_eligible == Some(true) => { accumulator.candidate_has_read_reserve() } - Some(candidate) => match inline_body_miss_reason_from_observations( + Some(candidate) => match verify_inline_body_from_observations( bucket.as_ref(), object.as_ref(), candidate, @@ -1386,8 +1390,11 @@ impl SetDisks { ) .await { - None => true, - Some(reason) => { + Ok(body) => { + verified_inline_body = Some(body); + true + } + Err(reason) => { final_miss_reason_override = Some(reason); if bounded_fanout && reason == GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD @@ -1445,6 +1452,7 @@ impl SetDisks { slots, diagnostics, condition_stopped, + verified_inline_body, }); } } @@ -1527,6 +1535,7 @@ impl SetDisks { slots, diagnostics, condition_stopped: false, + verified_inline_body: None, }) } diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 575dbb3e2..73fba73a3 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -939,6 +939,7 @@ struct OwnedGetObjectFileInfo { fi: FileInfo, parts_metadata: Vec, online_disks: Vec>, + verified_inline_body: Option, } enum LateMetadataFanout { @@ -955,6 +956,7 @@ impl GetObjectFileInfo { fi, parts_metadata, online_disks, + verified_inline_body: None, }), shared: None, condition_stopped: false, @@ -1017,6 +1019,27 @@ impl GetObjectFileInfo { self.late_metadata_fanout.is_some() } + fn set_verified_inline_body(&mut self, body: Option) { + if let Some(owned) = self.owned.as_mut() { + owned.verified_inline_body = body; + } + } + + fn take_verified_inline_body(&mut self) -> Option { + self.owned.as_mut().and_then(|owned| owned.verified_inline_body.take()) + } + + fn into_object_info(self, bucket: &str, object: &str, versioned: bool) -> ObjectInfo { + match (self.owned, self.shared) { + (Some(snapshot), None) => ObjectInfo::from_owned_file_info(snapshot.fi, bucket, object, versioned), + (None, Some(entry)) => match Arc::try_unwrap(entry) { + Ok(entry) => ObjectInfo::from_owned_file_info(entry.fi, bucket, object, versioned), + Err(entry) => ObjectInfo::from_file_info(&entry.fi, bucket, object, versioned), + }, + _ => unreachable!("GET metadata snapshot representation must be exclusive"), + } + } + fn into_owned(self) -> (FileInfo, Vec, Vec>) { let (fi, parts_metadata, online_disks, _) = self.into_owned_with_late_metadata_fanout(); (fi, parts_metadata, online_disks) @@ -2206,7 +2229,7 @@ mod prepared_get_object_metadata_tests { let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled(); rustfs_io_metrics::set_get_stage_metrics_enabled(true); - let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || { + let (restored, object_size, calls_total, body_decodes) = metrics::with_local_recorder(&recorder, || { runtime.block_on(async { let (_dirs, set_disks) = make_local_set_disks(4, 2).await; let opts = ObjectOptions { @@ -2260,7 +2283,12 @@ mod prepared_get_object_metadata_tests { .await .expect("inline GET body should stream"); - (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION)) + ( + restored, + object_size, + calls.total(disk_call_counters::KIND_READ_VERSION), + calls.total(disk_call_counters::KIND_BODY_DECODE), + ) }, ) .await @@ -2274,6 +2302,10 @@ mod prepared_get_object_metadata_tests { calls_total, 4, "default production inline GET should schedule the initial bounded quorum plus one hedge" ); + assert_eq!( + body_decodes, 1, + "cold inline early-stop GET must reuse the quorum-verified inline body instead of decoding shards twice" + ); assert_eq!( recorder.histogram_values( "rustfs_io_get_object_metadata_fanout_scheduled", @@ -2300,6 +2332,124 @@ mod prepared_get_object_metadata_tests { ); } + #[test] + #[serial_test::serial(body_cache_hook)] + fn get_object_info_uses_early_stop_for_locked_or_cache_safe_reads_and_full_wait_for_unlocked() { + use crate::storage_api_contracts::object::ObjectOperations; + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("current-thread runtime should build"); + let bucket = "get-object-info-early-stop-bucket"; + let object = object_with_initial_data_shards(bucket, "get-object-info-early-stop-object"); + let payload = b"head object early-stop payload".repeat(128); + + runtime.block_on(async { + let (_dirs, set_disks) = make_local_set_disks(4, 2).await; + let write_opts = ObjectOptions { + no_lock: true, + ..Default::default() + }; + + set_disks + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + let mut put_reader = PutObjReader::from_vec(payload.clone()); + set_disks + .put_object(bucket, &object, &mut put_reader, &write_opts) + .await + .expect("inline object should be written"); + + temp_env::async_with_vars( + [ + ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", None::<&str>), + ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>), + ], + async { + let slow_parity_disk = bounded_initial_parity_disk_index(bucket, &object); + let barrier = + rename_fanout_barrier::arm(&object, slow_parity_disk, rename_fanout_barrier::PHASE_READ_VERSION); + let calls = disk_call_counters::observe(&object); + let set_disks_for_head = Arc::clone(&set_disks); + let object_for_head = object.clone(); + let head_opts = ObjectOptions { + no_lock: true, + metadata_cache_safe: true, + ..Default::default() + }; + let mut head_task = + tokio::spawn( + async move { set_disks_for_head.get_object_info(bucket, &object_for_head, &head_opts).await }, + ); + + tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused()) + .await + .expect("cache-safe HEAD should pause a slow parity metadata read"); + let info = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut head_task) + .await + .expect("cache-safe HEAD should return before the paused parity metadata response") + .expect("HEAD task should not panic") + .expect("HEAD should succeed"); + assert_eq!(info.size, payload.len() as i64); + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 4, + "cache-safe HEAD should schedule bounded quorum plus one hedge" + ); + assert_eq!( + calls.total(disk_call_counters::KIND_BODY_DECODE), + 1, + "early-stop HEAD verifies inline quorum once" + ); + + // Second cache-safe HEAD should hit the QuorumSelected metadata cache entry (0 new disk reads). + let cached_info = set_disks + .get_object_info(bucket, &object, &ObjectOptions::default()) + .await + .expect("cached HEAD with default locked options should succeed"); + assert_eq!(cached_info.size, payload.len() as i64); + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 4, + "repeat locked HEAD must hit the QuorumSelected metadata cache entry" + ); + + // Release the slow parity barrier so full-wait can complete all 4 disks. + barrier.release(); + + // Unlocked internal preflight (no_lock=true, metadata_cache_safe=false) must bypass cache and full-wait. + let unlocked_info = set_disks + .get_object_info( + bucket, + &object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + ..Default::default() + }, + ) + .await + .expect("unlocked preflight get_object_info should succeed"); + assert_eq!(unlocked_info.size, payload.len() as i64); + assert_eq!( + calls.total(disk_call_counters::KIND_READ_VERSION), + 8, + "unlocked preflight get_object_info must query all 4 disks" + ); + assert_eq!( + calls.total(disk_call_counters::KIND_BODY_DECODE), + 1, + "unlocked preflight full-wait must not run early-stop inline body decode" + ); + }, + ) + .await; + }); + } + #[test] #[serial_test::serial(body_cache_hook)] fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index d40f10d19..1a26d4c0e 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -2873,7 +2873,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { }; let metadata_stage_start = stage_metrics_enabled.then(Instant::now); - let (snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() { + let (mut snapshot, prepared_object_info) = if let Some(prepared) = take_prepared_get_object_metadata() { (prepared.snapshot, prepared.object_info) } else { match self @@ -2903,6 +2903,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } } }; + let verified_inline_body = snapshot.take_verified_inline_body(); let fi = snapshot.fi(); let files = snapshot.parts_metadata(); let disks = snapshot.online_disks(); @@ -3010,6 +3011,42 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { disks.get(index).is_some_and(Option::is_some) }) { + if let Some(body) = verified_inline_body.filter(|body| body.len() == object_size) { + if let Some(inline_prepare_stage_start) = inline_prepare_stage_start.take() { + rustfs_io_metrics::record_get_object_stage_duration_by_size( + GET_OBJECT_PATH_INLINE_DIRECT, + GET_STAGE_INLINE_PREPARE, + object_class.as_str(), + size_bucket, + inline_prepare_stage_start.elapsed().as_secs_f64(), + ); + } + if rustfs_io_metrics::get_stage_metrics_enabled() { + rustfs_io_metrics::record_get_object_stage_duration_by_size( + GET_OBJECT_PATH_INLINE_DIRECT, + GET_STAGE_READER_SETUP, + object_class.as_str(), + size_bucket, + 0.0, + ); + rustfs_io_metrics::record_get_object_stage_duration_by_size( + GET_OBJECT_PATH_INLINE_DIRECT, + GET_STAGE_DECODE, + object_class.as_str(), + size_bucket, + 0.0, + ); + } + record_get_object_reader_path_observation(GET_OBJECT_PATH_INLINE_DIRECT, object_class, size_bucket); + let reader = GetObjectReader { + stream: Box::new(Cursor::new(body.clone())), + object_info, + buffered_body: Some(body), + body_source: GetObjectBodySource::Unprobed, + }; + return Ok(reader); + } + let read_length = inline_erasure_shard_file_offset( 0, object_size, @@ -10978,8 +11015,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // Use the same full xl.meta read path as GetObject metadata resolution. // This avoids HEAD/GetObject metadata visibility skew immediately after - // PutObject/CompleteMultipartUpload. - let read = self.get_object_fileinfo(bucket, object, opts, true, false); + // PutObject/CompleteMultipartUpload. Read-only callers that hold the + // shared lock and permit metadata caching can early-stop once write + // quorum agrees on the object identity. + let allow_early_stop = !opts.no_lock || opts.metadata_cache_safe; + let read = self.get_object_fileinfo(bucket, object, opts, true, allow_early_stop); #[cfg(feature = "diagnostic-read-version-caller")] let read = crate::cluster::rpc::runtime_sources::with_read_version_entry( rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ObjectInfoRead, @@ -10992,7 +11032,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { ); let snapshot = read.await.map_err(|e| to_object_err(e, vec![bucket, object]))?; - let oi = ObjectInfo::from_file_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended); + let oi = snapshot.into_object_info(bucket, object, opts.versioned || opts.version_suspended); Ok(oi) } diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index e5a96d57c..94e8b7f20 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -774,7 +774,7 @@ impl SetDisks { .await; #[cfg(not(feature = "diagnostic-read-version-caller"))] let metadata_read_result = metadata_read.await; - let metadata_read = match metadata_read_result { + let mut metadata_read = match metadata_read_result { Ok(metadata_read) => metadata_read, Err(err) => { let err = StorageError::from(err); @@ -784,6 +784,7 @@ impl SetDisks { }; let metadata_fanout_complete = metadata_read.is_complete(); let condition_stopped = metadata_read.condition_stopped; + let verified_inline_body = metadata_read.verified_inline_body.take(); let (mut parts_metadata, errs, metadata_fanout_diagnostics) = metadata_read.into_legacy(); metadata_fanout_diagnostics.record(metadata_metrics_path); // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata); @@ -902,6 +903,7 @@ impl SetDisks { GetObjectFileInfo::owned(fi, parts_metadata, op_online_disks) }; snapshot.condition_stopped = condition_stopped; + snapshot.set_verified_inline_body(verified_inline_body); Ok(snapshot) } @@ -916,7 +918,7 @@ impl SetDisks { Ok(snapshot) => snapshot, Err(e) => return (ObjectInfo::default(), FileInfo::default(), 0, Some(e)), }; - let fi = snapshot.fi().clone(); + let (fi, _, _) = snapshot.into_owned(); let write_quorum = fi.write_quorum(self.default_write_quorum());