opensource/rustfs
3
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 00:52:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): preserve omitted-disk GET recovery

Browse Source 
Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-13 00:27:12 +08:00
co-authored by heihutu zhi22915
parent 990aac45eb
commit 3999756d73
3 changed files with 89 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/mod.rs
+34 -25
View File
@@ -931,16 +931,21 @@ pub(in crate::set_disk) struct GetObjectFileInfo {
owned: Option<OwnedGetObjectFileInfo>,
shared: Option<Arc<GetObjectMetadataCacheEntry>>,
condition_stopped: bool,
late_metadata_fanout: Option<LateMetadataFanout>,
}
struct OwnedGetObjectFileInfo {
fi: FileInfo,
parts_metadata: Vec<FileInfo>,
online_disks: Vec<Option<DiskStore>>,
late_metadata_fanout_disks: Option<Vec<Option<DiskStore>>>,
}
type OwnedGetObjectFileInfoParts = (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>, Option<Vec<Option<DiskStore>>>);
enum LateMetadataFanout {
CapturedDisks(Vec<Option<DiskStore>>),
RefreshCurrentSet,
}
type OwnedGetObjectFileInfoParts = (FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>, Option<LateMetadataFanout>);
impl GetObjectFileInfo {
fn owned(fi: FileInfo, parts_metadata: Vec<FileInfo>, online_disks: Vec<Option<DiskStore>>) -> 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<Option<DiskStore>>,
late_metadata_fanout_disks: Vec<Option<DiskStore>>,
) -> 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<GetObjectMetadataCacheEntry>) -> 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<GetObjectMetadataCacheEntry>) -> 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<FileInfo>, Vec<Option<DiskStore>>) {
@@ -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"),
}
crates/ecstore/src/set_disk/ops/object.rs
+9 -5
View File
@@ -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,
crates/ecstore/src/set_disk/read.rs
+46 -3
View File
@@ -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::<Vec<_>>();
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;