diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index fad556f55..12713fd88 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -931,16 +931,21 @@ pub(in crate::set_disk) struct GetObjectFileInfo { owned: Option, shared: Option>, condition_stopped: bool, + late_metadata_fanout: Option, } struct OwnedGetObjectFileInfo { fi: FileInfo, parts_metadata: Vec, online_disks: Vec>, - late_metadata_fanout_disks: Option>>, } -type OwnedGetObjectFileInfoParts = (FileInfo, Vec, Vec>, Option>>); +enum LateMetadataFanout { + CapturedDisks(Vec>), + RefreshCurrentSet, +} + +type OwnedGetObjectFileInfoParts = (FileInfo, Vec, Vec>, Option); impl GetObjectFileInfo { fn owned(fi: FileInfo, parts_metadata: Vec, online_disks: Vec>) -> Self { @@ -949,10 +954,10 @@ impl GetObjectFileInfo { fi, parts_metadata, online_disks, - late_metadata_fanout_disks: None, }), shared: None, condition_stopped: false, + late_metadata_fanout: None, } } @@ -962,16 +967,9 @@ impl GetObjectFileInfo { online_disks: Vec>, late_metadata_fanout_disks: Vec>, ) -> Self { - Self { - owned: Some(OwnedGetObjectFileInfo { - fi, - parts_metadata, - online_disks, - late_metadata_fanout_disks: Some(late_metadata_fanout_disks), - }), - shared: None, - condition_stopped: false, - } + let mut snapshot = Self::owned(fi, parts_metadata, online_disks); + snapshot.late_metadata_fanout = Some(LateMetadataFanout::CapturedDisks(late_metadata_fanout_disks)); + snapshot } fn shared(entry: Arc) -> Self { @@ -980,9 +978,16 @@ impl GetObjectFileInfo { owned: None, shared: Some(entry), condition_stopped, + late_metadata_fanout: None, } } + fn shared_with_late_metadata_refresh(entry: Arc) -> Self { + let mut snapshot = Self::shared(entry); + snapshot.late_metadata_fanout = Some(LateMetadataFanout::RefreshCurrentSet); + snapshot + } + fn fi(&self) -> &FileInfo { match (&self.owned, &self.shared) { (Some(snapshot), None) => &snapshot.fi, @@ -1008,9 +1013,7 @@ impl GetObjectFileInfo { } fn has_late_metadata_fanout(&self) -> bool { - self.owned - .as_ref() - .is_some_and(|snapshot| snapshot.late_metadata_fanout_disks.is_some()) + self.late_metadata_fanout.is_some() } fn into_owned(self) -> (FileInfo, Vec, Vec>) { @@ -1019,16 +1022,22 @@ impl GetObjectFileInfo { } fn into_owned_with_late_metadata_fanout(self) -> OwnedGetObjectFileInfoParts { - match (self.owned, self.shared) { - (Some(snapshot), None) => ( - snapshot.fi, - snapshot.parts_metadata, - snapshot.online_disks, - snapshot.late_metadata_fanout_disks, - ), + let Self { + owned, + shared, + late_metadata_fanout, + .. + } = self; + match (owned, shared) { + (Some(snapshot), None) => (snapshot.fi, snapshot.parts_metadata, snapshot.online_disks, late_metadata_fanout), (None, Some(entry)) => match Arc::try_unwrap(entry) { - Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks, None), - Err(entry) => (entry.fi.clone(), entry.parts_metadata.clone(), entry.online_disks.clone(), None), + Ok(entry) => (entry.fi, entry.parts_metadata, entry.online_disks, late_metadata_fanout), + Err(entry) => ( + entry.fi.clone(), + entry.parts_metadata.clone(), + entry.online_disks.clone(), + late_metadata_fanout, + ), }, _ => unreachable!("GET metadata snapshot representation must be exclusive"), } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index a13252d8e..0365d0aa2 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -37,9 +37,9 @@ use super::super::{ GET_STAGE_LOCK_ACQUIRE, GET_STAGE_METADATA, GET_STAGE_OBJECT_INFO, GET_STAGE_PATH_DECISION, GET_STAGE_READER_SETUP, GenericError, GetCodecStreamingDecision, GetCodecStreamingFallbackReason, GetDirectMemoryDecision, GetObjectReader, HTTPRangeSpec, HashAlgorithm, HashMap, HashReader, HashSet, HeaderMap, HealChannelPriority, InstanceContext, Instant, - LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, ObjectKey, - ObjectLockConfigSnapshot, ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, Ordering, Pin, - PutObjReader, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReadPathPlan, ReaderImpl, ReplicateDecision, + LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_SET_DISK, LateMetadataFanout, OBJECT_OP_IGNORED_ERRS, ObjectApiError, ObjectInfo, + ObjectKey, ObjectLockConfigSnapshot, ObjectLockConfigState, ObjectOptions, ObjectReader, ObjectToDelete, OffsetDateTime, + Ordering, Pin, PutObjReader, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET, ReadPathPlan, ReaderImpl, ReplicateDecision, ReplicationObjectBridge, Result, SET_DISK_COMMIT_TAIL_WARN_THRESHOLD_MS, SLASH_SEPARATOR, SUFFIX_ACTUAL_SIZE, SUFFIX_COMPRESSION, SUFFIX_COMPRESSION_SIZE, SUFFIX_RESTORE_OPERATION_ID, SUFFIX_RESTORE_WORKER_LOCK, SetDisks, SmallWritePath, StorageError, TRANSITION_COMPLETE, UpdateMetadataOpts, Uuid, WriteLayout, adaptive_duplex_buffer_size, @@ -3197,9 +3197,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let object_size = usize::try_from(object_info.size) .map_err(|_| to_object_err(Error::other("two-phase GET object size is invalid"), vec![bucket, object]))?; let mut output = Vec::with_capacity(object_size); - let (fi, files, disks, late_metadata_fanout_disks) = snapshot.into_owned_with_late_metadata_fanout(); + let (fi, files, disks, late_metadata_fanout) = snapshot.into_owned_with_late_metadata_fanout(); let expected_identity = super::super::read::LateMetadataIdentity::from_file_info(&fi); - let late_metadata_fanout_disks = late_metadata_fanout_disks.ok_or_else(|| { + let late_metadata_fanout = late_metadata_fanout.ok_or_else(|| { to_object_err(Error::other("two-phase GET fallback context is missing"), vec![bucket, object]) })?; let initial_result = Self::get_object_with_fileinfo( @@ -3224,6 +3224,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { ) .await; if prepare_late_materialized_retry(&initial_result, &mut output, object_size) { + let late_metadata_fanout_disks = match late_metadata_fanout { + LateMetadataFanout::CapturedDisks(disks) => disks, + LateMetadataFanout::RefreshCurrentSet => self.get_disks_for_data_read().await, + }; #[cfg(test)] super::super::rename_fanout_barrier::checkpoint( object, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index ae017204f..b2d116309 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -653,7 +653,15 @@ impl SetDisks { GET_STAGE_METADATA_CACHE_LOOKUP, metadata_cache_lookup_start, ); - return Ok(GetObjectFileInfo::shared(cached)); + let snapshot = if cached.capability == GetObjectMetadataCacheCapability::QuorumSelected + && non_inline_data_read_early_stop_allowed(read_data, bucket, object) + && late_materialization_candidate_is_safe(&cached.fi) + { + GetObjectFileInfo::shared_with_late_metadata_refresh(cached) + } else { + GetObjectFileInfo::shared(cached) + }; + return Ok(snapshot); } MetadataCacheLookup::Hit(_) => { // Keep the conditional entry for other requests; this read @@ -4113,11 +4121,13 @@ mod metadata_cache_tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial(metadata_cache_early_stop)] async fn metadata_cache_publishes_quorum_selected_early_stop_result() { + use tokio::io::AsyncReadExt; + let isolated_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); isolated_ctx .update_erasure_type(crate::layout::endpoints::SetupType::Erasure) .await; - let (_dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(4, 2, isolated_ctx).await; + let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(4, 2, isolated_ctx).await; let bucket = "metadata-cache-early-stop"; let object = "objects/repeated-key"; let disks = set.disks.read().await.clone(); @@ -4169,7 +4179,40 @@ mod metadata_cache_tests { let cached = second.shared_entry().expect("second lookup should share cached metadata"); assert_eq!(cached.fi.name, object); - assert!(cached.online_disks.iter().filter(|disk| disk.is_some()).count() >= cached.read_quorum); + let cached_online_count = cached.online_disks.iter().filter(|disk| disk.is_some()).count(); + assert_eq!(cached_online_count, 3, "the warm cache should contain three of four disks"); + assert_eq!(cached.capability, GetObjectMetadataCacheCapability::QuorumSelected); + + let data_dir = cached + .fi + .data_dir + .expect("quorum cache retains the object data directory") + .to_string(); + let selected_disk_indices = cached + .online_disks + .iter() + .enumerate() + .filter_map(|(index, disk)| disk.as_ref().map(|_| index)) + .take(2) + .collect::>(); + assert_eq!(selected_disk_indices.len(), 2, "the warmed cache should name enough selected shards"); + for index in selected_disk_indices { + tokio::fs::remove_file(dirs[index].path().join(bucket).join(object).join(&data_dir).join("part.1")) + .await + .expect("simulate loss of two cached selected shards"); + } + + let mut reader = set + .get_object_reader(bucket, object, None, http::HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("a cached partial layout must recover through omitted disks"); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("the GET should decode after refreshing the omitted disks"); + assert_eq!(body, expected_body, "recovered body must be byte-exact"); }, ) .await;