mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
refactor(listing): list latest versions with MinIO's listing quorum (#8965)
* refactor(listing): list latest versions with MinIO's listing quorum Latest-version SetDisks::list_path now follows MinIO's er.listPath: ceil(asked/2) is the listing quorum for min_disks, directories and objects, agreed entries are emitted as read, and a split sample goes through MetaCacheEntries::resolve. One object can no longer fail a page. A set with fewer than half of its drives online still fails the page with ErasureReadQuorum. Remove the write-quorum layer this replaces: ListingSupplement, ListingReconciler and the LIST object read lock, the latest_listing_* quorum helpers, resolve_agreed_entry, resolve_with_write_quorum and the enforce_write_quorum / write_quorum_slack merge branches, and the per-version listing quorum helpers in filemeta. Make merge_file_meta_borrowed_streams pick a version id's most common header regardless of reader order, so an interrupted null-version overwrite no longer drops the committed generation from the merge. * test(listing): do not assert sampled optimal listings are deterministic optimal asks a random half of the drives, so an EC:8 object committed on nine of sixteen drives can miss the four-copy listing quorum in roughly one sample in six, exactly as in MinIO. Keep the deterministic assertion for strict and for optimal when every online drive is asked.
This commit is contained in:
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=347a0d4e5e547d5388703062ce66a66167a845b6e57ec591ccb4c732a567c68d
|
||||
sha256-linux=4f9efa47194279185bb68249e82c788fa147abe944b6c764050e3d47f31e7926
|
||||
sha256-darwin=3814cdde8cacf4187e18188d342ed70931cc32e268e42e5bb438d565bb4d0f81
|
||||
sha256-linux=663025a80273579f95ae3f49af02c0afcbfce94df09701c360a13e607b7637db
|
||||
|
||||
@@ -29,7 +29,7 @@
|
||||
{
|
||||
"invariant": "list-uncommitted-version",
|
||||
"suite": "rustfs-filemeta",
|
||||
"name": "metacache::tests::resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge"
|
||||
"name": "metacache::tests::resolve_merge_keeps_the_newest_version_at_listing_quorum"
|
||||
},
|
||||
{
|
||||
"invariant": "minio-object-fixture",
|
||||
|
||||
@@ -20,15 +20,14 @@
|
||||
//! offline before heal converges, a strict latest-listing quorum of 3 can only
|
||||
//! ever observe 2 copies, so ListObjectsV2 silently dropped the object even
|
||||
//! though GetObject (read quorum 2) still succeeded. Exposed by the flaky
|
||||
//! "Mixed-version rolling upgrade from rc.2" CI lane (run 33478999853); the
|
||||
//! product fix relaxes the listing's required object quorum by the number of
|
||||
//! set drives the listing could not consult (see
|
||||
//! `latest_listing_required_object_quorum` in
|
||||
//! `crates/ecstore/src/store/list_objects.rs`).
|
||||
//! "Mixed-version rolling upgrade from rc.2" CI lane (run 33478999853). The
|
||||
//! listing now follows MinIO's rule: an entry needs `ceil(asked/2)` copies
|
||||
//! among the online drives it asked, so three online drives need two copies
|
||||
//! (see `SetDisks::list_path` in `crates/ecstore/src/store/list_objects.rs`).
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::common::{RustFSTestClusterEnvironment, init_logging};
|
||||
use crate::common::{ClusterTopology, RustFSTestClusterEnvironment, init_logging};
|
||||
use aws_sdk_s3::Client;
|
||||
use bytes::Bytes;
|
||||
use std::collections::HashSet;
|
||||
@@ -51,13 +50,13 @@ mod tests {
|
||||
format!("degraded-object-{idx:02}")
|
||||
}
|
||||
|
||||
async fn list_all_keys(client: &Client) -> Result<HashSet<String>, Box<dyn Error + Send + Sync>> {
|
||||
async fn list_all_keys(client: &Client, bucket: &str) -> Result<HashSet<String>, Box<dyn Error + Send + Sync>> {
|
||||
let mut keys = HashSet::new();
|
||||
let mut continuation_token: Option<String> = None;
|
||||
loop {
|
||||
let response = client
|
||||
.list_objects_v2()
|
||||
.bucket(BUCKET)
|
||||
.bucket(bucket)
|
||||
.set_continuation_token(continuation_token.clone())
|
||||
.send()
|
||||
.await?;
|
||||
@@ -147,7 +146,7 @@ mod tests {
|
||||
let expected: HashSet<String> = (0..OBJECT_COUNT).map(object_key).collect();
|
||||
let deadline = Instant::now() + LISTING_DEADLINE;
|
||||
let listed = loop {
|
||||
let listed = match list_all_keys(&client).await {
|
||||
let listed = match list_all_keys(&client, BUCKET).await {
|
||||
Ok(keys) => keys,
|
||||
Err(error) if Instant::now() < deadline => {
|
||||
info!("retrying degraded listing: {error}");
|
||||
@@ -171,4 +170,97 @@ mod tests {
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 4-node x 4-drive cluster (one 16-drive set), #8828 shape:
|
||||
/// 1. PUT objects while the storage class is EC:8, so each is 8+8.
|
||||
/// 2. Restart with the EC:4 default, as after a storage-class change.
|
||||
/// 3. Stop two nodes (8 drives). Each object keeps eight data shards, so
|
||||
/// GET succeeds and ListObjectsV2 must list every key: the listing
|
||||
/// quorum is half of the eight drives it can ask.
|
||||
/// 4. Stop a third node. Four online drives are below the set's read
|
||||
/// quorum, so LIST must report 503 instead of a partial page.
|
||||
#[tokio::test]
|
||||
async fn ec8_objects_stay_listable_with_half_of_an_ec4_set_offline() -> TestResult {
|
||||
const EC8_BUCKET: &str = "degraded-listing-ec8";
|
||||
init_logging();
|
||||
|
||||
let mut cluster = RustFSTestClusterEnvironment::with_topology(ClusterTopology::single_pool_multidrive(4, 4)).await?;
|
||||
cluster.set_env("RUSTFS_HEAL_ENABLED", "false");
|
||||
cluster.set_env("RUSTFS_SCANNER_ENABLED", "false");
|
||||
cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:8");
|
||||
cluster.start().await?;
|
||||
cluster.create_test_bucket(EC8_BUCKET).await?;
|
||||
let client = cluster.create_s3_client(0)?;
|
||||
for idx in 0..OBJECT_COUNT {
|
||||
let key = object_key(idx);
|
||||
client
|
||||
.put_object()
|
||||
.bucket(EC8_BUCKET)
|
||||
.key(&key)
|
||||
.body(Bytes::from(format!("ec8 listing payload {idx}")).into())
|
||||
.send()
|
||||
.await
|
||||
.map_err(|error| format!("EC:8 PUT for {key} failed: {error}"))?;
|
||||
}
|
||||
|
||||
info!("restarting with the EC:4 default storage class");
|
||||
cluster.stop();
|
||||
// A later entry for the same variable wins when the node process starts.
|
||||
cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:4");
|
||||
cluster.start().await?;
|
||||
let client = cluster.create_s3_client(0)?;
|
||||
|
||||
info!("stopping nodes 2 and 3 (eight of sixteen drives)");
|
||||
cluster.stop_node(2)?;
|
||||
cluster.stop_node(3)?;
|
||||
for idx in 0..OBJECT_COUNT {
|
||||
let key = object_key(idx);
|
||||
let deadline = Instant::now() + GET_RETRY_DEADLINE;
|
||||
loop {
|
||||
match client.get_object().bucket(EC8_BUCKET).key(&key).send().await {
|
||||
Ok(response) => {
|
||||
let body = response.body.collect().await?.into_bytes();
|
||||
assert_eq!(body.as_ref(), format!("ec8 listing payload {idx}").as_bytes());
|
||||
break;
|
||||
}
|
||||
Err(error) if Instant::now() < deadline => {
|
||||
info!("retrying EC:8 GET for {key}: {error}");
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
Err(error) => return Err(format!("EC:8 object {key} failed GET with eight drives: {error}").into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let expected: HashSet<String> = (0..OBJECT_COUNT).map(object_key).collect();
|
||||
let deadline = Instant::now() + LISTING_DEADLINE;
|
||||
loop {
|
||||
match list_all_keys(&client, EC8_BUCKET).await {
|
||||
Ok(listed) if listed == expected => break,
|
||||
Ok(listed) => assert!(
|
||||
Instant::now() < deadline,
|
||||
"EC:8 objects readable with eight drives stayed missing from ListObjectsV2: listed={listed:?}"
|
||||
),
|
||||
Err(error) => assert!(
|
||||
Instant::now() < deadline,
|
||||
"ListObjectsV2 kept failing with eight of sixteen drives online: {error}"
|
||||
),
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
|
||||
info!("stopping node 1 (four of sixteen drives remain)");
|
||||
cluster.stop_node(1)?;
|
||||
let deadline = Instant::now() + LISTING_DEADLINE;
|
||||
loop {
|
||||
match client.list_objects_v2().bucket(EC8_BUCKET).send().await {
|
||||
Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => break,
|
||||
Err(error) => assert!(Instant::now() < deadline, "below the set read quorum LIST must report 503, got {error}"),
|
||||
Ok(response) => panic!("below the set read quorum LIST must fail, but listed {} keys", response.contents().len()),
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -43,11 +43,6 @@ const EVENT_METACACHE_LISTING: &str = "metacache_listing";
|
||||
pub type AgreedFn = Box<dyn Fn(MetaCacheEntry) -> Pin<Box<dyn Future<Output = Result<(), DiskError>> + Send>> + Send + 'static>;
|
||||
pub type PartialFn =
|
||||
Box<dyn Fn(MetaCacheEntries, &[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
|
||||
pub(crate) type PartialResultFn = Box<
|
||||
dyn Fn(MetaCacheEntries, &[Option<DiskError>]) -> Pin<Box<dyn Future<Output = disk::error::Result<()>> + Send>>
|
||||
+ Send
|
||||
+ 'static,
|
||||
>;
|
||||
type FinishedFn = Box<dyn Fn(&[Option<DiskError>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
@@ -60,6 +55,7 @@ impl FallbackClaimTracker {
|
||||
self.claimed.lock().await.insert(disk.endpoint().to_string());
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) async fn claimed_keys(&self) -> HashSet<String> {
|
||||
self.claimed.lock().await.clone()
|
||||
}
|
||||
@@ -317,7 +313,17 @@ pub struct ListPathRawOutcome {
|
||||
pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> {
|
||||
let rx = rx.child_token();
|
||||
let _cancel_guard = rx.clone().drop_guard();
|
||||
list_path_raw_inner(rx, opts, None, None).await.map(|_| ())
|
||||
list_path_raw_inner(rx, opts, None).await.map(|_| ())
|
||||
}
|
||||
|
||||
/// Like [`list_path_raw`], but reports how the merged stream ended.
|
||||
pub(crate) async fn list_path_raw_with_outcome(
|
||||
rx: CancellationToken,
|
||||
opts: ListPathRawOptions,
|
||||
) -> disk::error::Result<ListPathRawOutcome> {
|
||||
let rx = rx.child_token();
|
||||
let _cancel_guard = rx.clone().drop_guard();
|
||||
list_path_raw_inner(rx, opts, None).await
|
||||
}
|
||||
|
||||
pub(crate) async fn list_path_raw_with_claim_tracker(
|
||||
@@ -327,25 +333,13 @@ pub(crate) async fn list_path_raw_with_claim_tracker(
|
||||
) -> disk::error::Result<ListPathRawOutcome> {
|
||||
let rx = rx.child_token();
|
||||
let _cancel_guard = rx.clone().drop_guard();
|
||||
list_path_raw_inner(rx, opts, Some(claim_tracker), None).await
|
||||
}
|
||||
|
||||
pub(crate) async fn list_path_raw_with_partial_result(
|
||||
rx: CancellationToken,
|
||||
opts: ListPathRawOptions,
|
||||
claim_tracker: FallbackClaimTracker,
|
||||
partial_result: PartialResultFn,
|
||||
) -> disk::error::Result<ListPathRawOutcome> {
|
||||
let rx = rx.child_token();
|
||||
let _cancel_guard = rx.clone().drop_guard();
|
||||
list_path_raw_inner(rx, opts, Some(claim_tracker), Some(partial_result)).await
|
||||
list_path_raw_inner(rx, opts, Some(claim_tracker)).await
|
||||
}
|
||||
|
||||
async fn list_path_raw_inner(
|
||||
rx: CancellationToken,
|
||||
opts: ListPathRawOptions,
|
||||
fallback_claim_tracker: Option<FallbackClaimTracker>,
|
||||
partial_result: Option<PartialResultFn>,
|
||||
) -> disk::error::Result<ListPathRawOutcome> {
|
||||
if opts.disks.is_empty() {
|
||||
return Err(DiskError::ErasureReadQuorum);
|
||||
@@ -1020,13 +1014,7 @@ async fn list_path_raw_inner(
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(partial_fn) = partial_result.as_ref() {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = revjob_rx.cancelled() => return Ok(ListPathRawOutcome { limit_reached }),
|
||||
result = partial_fn(MetaCacheEntries(top_entries), &errs) => result?,
|
||||
}
|
||||
} else if let Some(partial_fn) = opts.partial.as_ref() {
|
||||
if let Some(partial_fn) = opts.partial.as_ref() {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = revjob_rx.cancelled() => return Ok(ListPathRawOutcome { limit_reached }),
|
||||
@@ -1308,7 +1296,6 @@ mod tests {
|
||||
..Default::default()
|
||||
},
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("listing should complete");
|
||||
@@ -1333,7 +1320,6 @@ mod tests {
|
||||
..Default::default()
|
||||
},
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("listing should complete");
|
||||
@@ -1827,39 +1813,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_path_raw_propagates_partial_callback_failure() {
|
||||
let first = MetaCacheEntry {
|
||||
name: "bucket/object-a".to_string(),
|
||||
metadata: vec![1],
|
||||
..Default::default()
|
||||
};
|
||||
let second = MetaCacheEntry {
|
||||
name: "bucket/object-b".to_string(),
|
||||
metadata: vec![2],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let err = list_path_raw_with_partial_result(
|
||||
CancellationToken::new(),
|
||||
ListPathRawOptions {
|
||||
disks: vec![None, None],
|
||||
min_disks: 2,
|
||||
test_reader_behaviors: vec![
|
||||
TestReaderBehavior::Entries(vec![first]),
|
||||
TestReaderBehavior::Entries(vec![second]),
|
||||
],
|
||||
..Default::default()
|
||||
},
|
||||
FallbackClaimTracker::default(),
|
||||
Box::new(move |_, _| Box::pin(async { Err(DiskError::ErasureReadQuorum) })),
|
||||
)
|
||||
.await
|
||||
.expect_err("a resolution failure after reader advancement must fail the listing");
|
||||
|
||||
assert_eq!(err, DiskError::ErasureReadQuorum);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_path_raw_cancels_a_blocked_partial_callback() {
|
||||
let cancel = CancellationToken::new();
|
||||
|
||||
@@ -160,7 +160,7 @@ use rustfs_utils::http::{
|
||||
use rustfs_utils::{
|
||||
HashAlgorithm,
|
||||
crypto::hex,
|
||||
path::{SLASH_SEPARATOR, encode_dir_object, encode_dir_object_ref, has_suffix, object_name_disk_identity, path_join_buf},
|
||||
path::{SLASH_SEPARATOR, encode_dir_object, has_suffix, object_name_disk_identity, path_join_buf},
|
||||
};
|
||||
use sha2::Sha256;
|
||||
use std::borrow::Cow;
|
||||
@@ -5247,40 +5247,6 @@ impl SetDisks {
|
||||
.with_attempt(attempt))
|
||||
}
|
||||
|
||||
pub(crate) async fn read_listing_metadata_after_namespace_barrier(
|
||||
&self,
|
||||
disks: &[Option<DiskStore>],
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
) -> Result<(MetaCacheEntries, usize)> {
|
||||
// This is only called after the lock-free LIST fast path observes an
|
||||
// unresolved metadata generation. Waiting on the ordinary object read
|
||||
// lock establishes a publication boundary with concurrent overwrites.
|
||||
let stored_object = encode_dir_object_ref(object);
|
||||
let _guard = self
|
||||
.acquire_read_lock_diag("list_object_reconcile", bucket, stored_object.as_ref())
|
||||
.await?;
|
||||
let (raw_entries, errors) = Self::read_all_raw_file_info(disks, bucket, stored_object.as_ref(), false).await;
|
||||
let confirmed_absent = errors
|
||||
.iter()
|
||||
.flatten()
|
||||
.filter(|err| DiskError::is_err_object_not_found(err) || DiskError::is_err_version_not_found(err))
|
||||
.count();
|
||||
let entries = raw_entries
|
||||
.into_iter()
|
||||
.map(|raw| {
|
||||
raw.filter(|raw| !raw.buf.is_empty()).map(|raw| MetaCacheEntry {
|
||||
name: object.to_owned(),
|
||||
metadata: raw.buf,
|
||||
cached: None,
|
||||
reusable: false,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok((MetaCacheEntries(entries), confirmed_absent))
|
||||
}
|
||||
|
||||
async fn acquire_write_lock_diag(&self, op: &'static str, bucket: &str, object: &str) -> Result<ObjectLockDiagGuard> {
|
||||
crate::hp_guard!("SetDisks::acquire_write_lock");
|
||||
let diag_enabled = is_object_lock_diag_enabled();
|
||||
@@ -16477,54 +16443,6 @@ mod tests {
|
||||
assert_eq!(err, StorageError::PreconditionFailed);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn listing_metadata_reconcile_encodes_directory_object_and_waits_for_writer() {
|
||||
let set_disks = make_local_bucket_test_set_disks().await;
|
||||
let bucket = "bucket-list-reconcile-lock";
|
||||
let object = "directory/";
|
||||
let stored_object = encode_dir_object(object);
|
||||
set_disks
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("bucket should be created");
|
||||
let mut reader = PutObjReader::from_vec(b"directory marker".to_vec());
|
||||
set_disks
|
||||
.put_object(bucket, &stored_object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("directory object should be written under its encoded key");
|
||||
let disks = set_disks.disks.read().await.clone();
|
||||
let namespace_lock = set_disks
|
||||
.new_ns_lock(bucket, &stored_object)
|
||||
.await
|
||||
.expect("namespace lock should be created");
|
||||
let writer_guard = namespace_lock
|
||||
.get_write_lock(std::time::Duration::from_secs(30))
|
||||
.await
|
||||
.expect("writer lock should be acquired");
|
||||
let read = set_disks.read_listing_metadata_after_namespace_barrier(&disks, bucket, object);
|
||||
tokio::pin!(read);
|
||||
assert!(
|
||||
tokio::time::timeout(std::time::Duration::from_millis(50), &mut read)
|
||||
.await
|
||||
.is_err(),
|
||||
"LIST reconciliation must wait for the publishing writer"
|
||||
);
|
||||
drop(writer_guard);
|
||||
|
||||
let (entries, confirmed_absent) = tokio::time::timeout(std::time::Duration::from_secs(30), read)
|
||||
.await
|
||||
.expect("LIST reconciliation should resume after writer release")
|
||||
.expect("LIST reconciliation should read after the namespace barrier");
|
||||
assert_eq!(entries.0.len(), disks.len());
|
||||
assert!(
|
||||
entries.0.iter().all(|entry| entry
|
||||
.as_ref()
|
||||
.is_some_and(|entry| entry.name == object && !entry.metadata.is_empty())),
|
||||
"LIST reconciliation must read encoded metadata and retain the logical key"
|
||||
);
|
||||
assert_eq!(confirmed_absent, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn repeated_body_write_keeps_etag_but_changes_data_dir_generation() {
|
||||
let set_disks = make_local_bucket_test_set_disks().await;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -14,14 +14,14 @@
|
||||
|
||||
use super::*;
|
||||
use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
|
||||
use crate::disk::STORAGE_FORMAT_FILE;
|
||||
use crate::disk::{DiskAPI, STORAGE_FORMAT_FILE, WalkDirOptions};
|
||||
use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
|
||||
use bytes::Bytes;
|
||||
use rustfs_filemeta::{FileInfo, ObjectPartInfo};
|
||||
use rustfs_filemeta::{FileInfo, MetacacheReader, ObjectPartInfo};
|
||||
use std::collections::BTreeMap;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use tokio::io::{AsyncWrite, DuplexStream};
|
||||
use tokio::io::{AsyncWrite, DuplexStream, duplex};
|
||||
use tokio::sync::Notify;
|
||||
use tokio::time::timeout;
|
||||
|
||||
@@ -468,6 +468,50 @@ async fn list_objects_shared_corpus_offline_corrupt_and_minority_stale() {
|
||||
fixture.assert_walk("", None, None, 3).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn list_objects_shared_corpus_quorum_delete_leaves_no_resurrectable_stale_copy() {
|
||||
let mut fixture = Fixture::new("list-corpus-stale-copies", namespace(&shared_keys()), true).await;
|
||||
// A delete that reached write quorum (three of four drives) leaves one
|
||||
// stale copy, the most any quorum delete can leave on this set.
|
||||
let deleted = ["a0", "z"];
|
||||
for key in deleted {
|
||||
let physical_key = rustfs_utils::path::encode_dir_object(key);
|
||||
for dir in fixture.dirs.iter().take(3) {
|
||||
tokio::fs::remove_dir_all(dir.path().join(&fixture.bucket).join(&physical_key))
|
||||
.await
|
||||
.expect("quorum delete fixture should remove the object directory");
|
||||
}
|
||||
}
|
||||
fixture.objects.retain(|object| !deleted.contains(&object.key.as_str()));
|
||||
|
||||
temp_env::async_with_vars([(ENV_API_LIST_OBJECTS_QUORUM, Some("strict"))], async {
|
||||
for max in [1, 3, 1000] {
|
||||
fixture.assert_walk("", None, None, max).await;
|
||||
fixture.assert_walk("", Some("/"), None, max).await;
|
||||
}
|
||||
})
|
||||
.await;
|
||||
|
||||
// A sampling mode promises only that no committed key goes missing.
|
||||
temp_env::async_with_vars([(ENV_API_LIST_OBJECTS_QUORUM, Some("optimal"))], async {
|
||||
let mut listed = Vec::new();
|
||||
let mut marker = None;
|
||||
for _ in 0..=fixture.objects.len() + deleted.len() {
|
||||
let page = fixture.page("", marker.clone(), None, 3).await;
|
||||
listed.extend(page.objects.into_iter().map(|object| object.name));
|
||||
if !page.is_truncated {
|
||||
break;
|
||||
}
|
||||
marker = page.next_marker;
|
||||
}
|
||||
for object in &fixture.objects {
|
||||
assert!(listed.contains(&object.key), "optimal LIST lost committed key {}", object.key);
|
||||
}
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_objects_shared_corpus_versions_delete_markers_and_free_versions() {
|
||||
let mut objects = vec![
|
||||
|
||||
@@ -446,7 +446,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: false,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
@@ -463,24 +462,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: false,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
group.bench_function("8_disks_write_quorum/1000_objects", |b| {
|
||||
b.iter(|| {
|
||||
for entries in &sets_8_agree {
|
||||
let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
|
||||
dir_quorum: 4,
|
||||
obj_quorum: 4,
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: true,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
@@ -491,13 +472,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
group.bench_function("8_disks_disagree_merge/1000_objects", |b| {
|
||||
b.iter(|| {
|
||||
for entries in &sets_8_disagree {
|
||||
let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
|
||||
let resolved = entries.resolve(MetadataResolutionParams {
|
||||
dir_quorum: 4,
|
||||
obj_quorum: 4,
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: true,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
@@ -508,13 +488,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
group.bench_function("8_disks_agree_13_versions/1000_objects", |b| {
|
||||
b.iter(|| {
|
||||
for entries in &sets_8_agree_13v {
|
||||
let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
|
||||
let resolved = entries.resolve(MetadataResolutionParams {
|
||||
dir_quorum: 4,
|
||||
obj_quorum: 4,
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: true,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
@@ -525,13 +504,12 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
group.bench_function("8_disks_disagree_13_versions/1000_objects", |b| {
|
||||
b.iter(|| {
|
||||
for entries in &sets_8_disagree_13v {
|
||||
let resolved = entries.resolve_with_write_quorum(MetadataResolutionParams {
|
||||
let resolved = entries.resolve(MetadataResolutionParams {
|
||||
dir_quorum: 4,
|
||||
obj_quorum: 4,
|
||||
requested_versions: 1,
|
||||
bucket: "bucket".to_string(),
|
||||
strict: true,
|
||||
write_quorum_slack: 0,
|
||||
candidates: Vec::new(),
|
||||
});
|
||||
black_box(resolved);
|
||||
@@ -596,191 +574,6 @@ fn bench_metacache_resolve(c: &mut Criterion) {
|
||||
group.finish();
|
||||
}
|
||||
|
||||
fn listing_history(versions: usize, cached: bool) -> MetaCacheEntry {
|
||||
let mut metadata = FileMeta::new();
|
||||
for index in 0..versions {
|
||||
let (data, parity) = if index + 1 == versions { (12, 4) } else { (8, 8) };
|
||||
let mut info = FileInfo::new("object", data, parity);
|
||||
info.volume = "bucket".to_owned();
|
||||
info.erasure.index = 1;
|
||||
info.data_dir = Some(Uuid::from_u128(0x1000 + u128::try_from(index).expect("data directory index")));
|
||||
info.versioned = true;
|
||||
info.version_id = Some(Uuid::from_u128(u128::try_from(index + 1).expect("version index")));
|
||||
info.mod_time = Some(
|
||||
OffsetDateTime::from_unix_timestamp(1_700_000_000 + i64::try_from(index).expect("timestamp")).expect("fixture time"),
|
||||
);
|
||||
info.size = 1;
|
||||
info.parts = vec![ObjectPartInfo {
|
||||
number: 1,
|
||||
size: 1,
|
||||
actual_size: 1,
|
||||
..Default::default()
|
||||
}];
|
||||
info.metadata.insert("etag".to_owned(), format!("etag-{index}"));
|
||||
info.metadata.insert("x-amz-meta-description".to_owned(), "m".repeat(1024));
|
||||
metadata.add_version(info).expect("valid version history");
|
||||
}
|
||||
MetaCacheEntry {
|
||||
name: "object".to_owned(),
|
||||
metadata: metadata.marshal_msg().expect("serialize history"),
|
||||
cached: cached.then_some(metadata),
|
||||
reusable: false,
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn owned_agreed_latest_reference(
|
||||
mut entry: MetaCacheEntry,
|
||||
replicas: usize,
|
||||
params: &rustfs_filemeta::MetadataResolutionParams,
|
||||
enforce_write_quorum: bool,
|
||||
) -> rustfs_filemeta::MetaCacheEntryResolution {
|
||||
use rustfs_filemeta::MetaCacheEntryResolution::{NeedsSupplement, Rejected, Resolved};
|
||||
if !enforce_write_quorum {
|
||||
return Resolved(entry);
|
||||
}
|
||||
if entry.is_dir() {
|
||||
return if replicas >= params.dir_quorum {
|
||||
Resolved(entry)
|
||||
} else {
|
||||
Rejected
|
||||
};
|
||||
}
|
||||
if replicas < params.obj_quorum {
|
||||
return Rejected;
|
||||
}
|
||||
let metadata = match entry.xl_meta_ref() {
|
||||
Ok(metadata) => metadata,
|
||||
Err(_) if params.obj_quorum <= 1 => return Resolved(entry),
|
||||
Err(_) => return Rejected,
|
||||
};
|
||||
if params.requested_versions == 1
|
||||
&& let Some(first) = metadata.versions.first()
|
||||
&& !first.header.free_version()
|
||||
&& replicas >= first.listing_quorum(params.obj_quorum, params.write_quorum_slack)
|
||||
{
|
||||
return Resolved(entry);
|
||||
}
|
||||
// These fixtures cover the fast prefix, never older-version reconciliation.
|
||||
NeedsSupplement(entry)
|
||||
}
|
||||
|
||||
fn bench_listing_borrowed_metadata(c: &mut Criterion) {
|
||||
use rustfs_filemeta::{FileMetaVersion, MetaCacheEntries, MetadataResolutionParams};
|
||||
let mut group = c.benchmark_group("listing_borrowed_metadata");
|
||||
group
|
||||
.sample_size(20)
|
||||
.warm_up_time(Duration::from_millis(150))
|
||||
.measurement_time(Duration::from_millis(700));
|
||||
for versions in [2, 16, 256] {
|
||||
for cached in [false, true] {
|
||||
let entry = listing_history(versions, cached);
|
||||
let params = MetadataResolutionParams {
|
||||
dir_quorum: 8,
|
||||
obj_quorum: 8,
|
||||
requested_versions: 1,
|
||||
write_quorum_slack: 8,
|
||||
bucket: "bucket".to_owned(),
|
||||
..Default::default()
|
||||
};
|
||||
let slots = vec![Some(&entry); 8];
|
||||
let owned = MetaCacheEntries(vec![Some(entry.clone()); 8])
|
||||
.resolve_with_write_quorum(params.clone())
|
||||
.expect("previous committed version");
|
||||
let borrowed =
|
||||
MetaCacheEntries::resolve_borrowed_with_write_quorum(&slots, params.clone()).expect("borrowed previous version");
|
||||
assert_eq!(owned.metadata, borrowed.metadata);
|
||||
let id = format!("{versions}_versions/{}", if cached { "cached" } else { "cold" });
|
||||
group.throughput(Throughput::Bytes(u64::try_from(entry.metadata.len()).expect("metadata size")));
|
||||
// ABBA order permits an explicit baseline-drift check for each fixture.
|
||||
for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
|
||||
group.bench_function(BenchmarkId::new("replicas", format!("{id}/{phase}")), |b| {
|
||||
b.iter(|| {
|
||||
if phase.starts_with("owned") {
|
||||
black_box(
|
||||
MetaCacheEntries(vec![Some(black_box(&entry).clone()); 8])
|
||||
.resolve_with_write_quorum(params.clone()),
|
||||
)
|
||||
} else {
|
||||
let slots = vec![Some(black_box(&entry)); 8];
|
||||
black_box(MetaCacheEntries::resolve_borrowed_with_write_quorum(&slots, params.clone()))
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
let mut agreed_metadata = entry
|
||||
.cached
|
||||
.clone()
|
||||
.unwrap_or_else(|| FileMeta::load(&entry.metadata).expect("agreed fixture"));
|
||||
let mut latest = FileMetaVersion::try_from(agreed_metadata.versions[0].meta.as_slice()).expect("latest body");
|
||||
let object = latest.object.as_mut().expect("latest object");
|
||||
object.erasure_m = 8;
|
||||
object.erasure_n = 8;
|
||||
agreed_metadata.versions[0] = latest.try_into().expect("balanced latest header");
|
||||
let agreed = MetaCacheEntry {
|
||||
name: entry.name.clone(),
|
||||
metadata: agreed_metadata.marshal_msg().expect("agreed history"),
|
||||
cached: cached.then_some(agreed_metadata),
|
||||
reusable: false,
|
||||
};
|
||||
for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
|
||||
group.bench_function(BenchmarkId::new("agreed", format!("{id}/{phase}")), |b| {
|
||||
b.iter_batched(
|
||||
|| agreed.clone(),
|
||||
|candidate| {
|
||||
let resolved = if phase.starts_with("owned") {
|
||||
owned_agreed_latest_reference(candidate, 8, ¶ms, true)
|
||||
} else {
|
||||
rustfs_filemeta::resolve_agreed_entry(candidate, 8, ¶ms, true)
|
||||
};
|
||||
let rustfs_filemeta::MetaCacheEntryResolution::Resolved(candidate) = resolved else {
|
||||
panic!("balanced latest must use agreed fast prefix");
|
||||
};
|
||||
black_box(candidate)
|
||||
},
|
||||
BatchSize::SmallInput,
|
||||
);
|
||||
});
|
||||
}
|
||||
let replicas: Vec<_> = (1..=8)
|
||||
.map(|disk| {
|
||||
let mut metadata = entry
|
||||
.cached
|
||||
.clone()
|
||||
.unwrap_or_else(|| FileMeta::load(&entry.metadata).expect("decode fixture"));
|
||||
for version in &mut metadata.versions {
|
||||
let mut payload = FileMetaVersion::try_from(version.meta.as_slice()).expect("version body");
|
||||
payload.object.as_mut().expect("object body").erasure_index = disk;
|
||||
version.meta = payload.marshal_msg().expect("per-disk body");
|
||||
}
|
||||
MetaCacheEntry {
|
||||
name: entry.name.clone(),
|
||||
metadata: metadata.marshal_msg().expect("per-disk metadata"),
|
||||
cached: cached.then_some(metadata),
|
||||
reusable: false,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
assert!(replicas[0].headers_agree(&replicas, false));
|
||||
for phase in ["owned_a1", "borrowed_b1", "borrowed_b2", "owned_a2"] {
|
||||
group.bench_function(BenchmarkId::new("headers", format!("{id}/{phase}")), |b| {
|
||||
b.iter(|| {
|
||||
let agreed = if phase.starts_with("owned") {
|
||||
replicas
|
||||
.iter()
|
||||
.all(|candidate| replicas[0].matches(Some(black_box(candidate)), false).1)
|
||||
} else {
|
||||
replicas[0].headers_agree(black_box(&replicas), false)
|
||||
};
|
||||
black_box(agreed)
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(
|
||||
benches,
|
||||
bench_create_real_xlmeta,
|
||||
@@ -797,8 +590,7 @@ criterion_group!(
|
||||
bench_list_page_to_fileinfo,
|
||||
bench_delete_inline_versions,
|
||||
bench_metacache_stream_page,
|
||||
bench_metacache_resolve,
|
||||
bench_listing_borrowed_metadata
|
||||
bench_metacache_resolve
|
||||
);
|
||||
|
||||
criterion_main!(benches);
|
||||
|
||||
@@ -155,7 +155,6 @@ pub fn is_skip_meta_key(key: &str) -> bool {
|
||||
}
|
||||
|
||||
mod codec;
|
||||
pub(crate) use codec::{BorrowedFileMeta, reconcile_headers};
|
||||
mod inline_data;
|
||||
pub(crate) mod msgp_decode;
|
||||
mod validation;
|
||||
|
||||
@@ -622,18 +622,6 @@ pub struct FileMetaShallowVersion {
|
||||
pub meta: Vec<u8>, // FileMetaVersion.marshal_msg
|
||||
}
|
||||
|
||||
fn write_quorum_from_erasure(data_blocks: usize, parity_blocks: usize) -> Option<usize> {
|
||||
if data_blocks == 0 {
|
||||
return None;
|
||||
}
|
||||
|
||||
Some(if data_blocks == parity_blocks {
|
||||
data_blocks.saturating_add(1)
|
||||
} else {
|
||||
data_blocks
|
||||
})
|
||||
}
|
||||
|
||||
impl FileMetaShallowVersion {
|
||||
/// Parse version meta with legacy format compatibility.
|
||||
/// Use this instead of `FileMetaVersion::default()` + `unmarshal_msg()` to handle old-version xl.meta.
|
||||
@@ -641,60 +629,11 @@ impl FileMetaShallowVersion {
|
||||
FileMetaVersion::try_from(self.meta.as_slice())
|
||||
}
|
||||
|
||||
pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
|
||||
write_quorum_from_header_and_meta(&self.header, self.meta.as_slice(), fallback_quorum)
|
||||
}
|
||||
|
||||
/// Quorum for listing this version without accepting an unreadable payload.
|
||||
pub fn listing_quorum(&self, fallback_quorum: usize, unreachable_disks: usize) -> usize {
|
||||
listing_quorum_from_header_and_meta(&self.header, &self.meta, fallback_quorum, unreachable_disks)
|
||||
}
|
||||
|
||||
pub fn into_fileinfo(&self, volume: &str, path: &str, all_parts: bool) -> Result<FileInfo> {
|
||||
FileMetaVersion::decode_fileinfo_from_slice(self.meta.as_slice(), volume, path, all_parts, true)
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub(crate) fn write_quorum_from_header_and_meta(header: &FileMetaVersionHeader, meta: &[u8], fallback_quorum: usize) -> usize {
|
||||
let header_quorum = header.write_quorum(fallback_quorum);
|
||||
if header.has_ec() || !matches!(header.version_type, VersionType::Object | VersionType::Legacy) {
|
||||
return header_quorum;
|
||||
}
|
||||
|
||||
FileMetaVersion::try_from(meta)
|
||||
.ok()
|
||||
.and_then(|version| version.erasure_write_quorum())
|
||||
.unwrap_or(header_quorum)
|
||||
}
|
||||
|
||||
pub(crate) fn listing_quorum_from_header_and_meta(
|
||||
header: &FileMetaVersionHeader,
|
||||
meta: &[u8],
|
||||
fallback_quorum: usize,
|
||||
unreachable_disks: usize,
|
||||
) -> usize {
|
||||
let geometry = if header.version_type == VersionType::Object && header.has_ec() {
|
||||
Some((usize::from(header.ec_m), usize::from(header.ec_n)))
|
||||
} else if matches!(header.version_type, VersionType::Object | VersionType::Legacy) {
|
||||
FileMetaVersion::try_from(meta)
|
||||
.ok()
|
||||
.and_then(|version| version.erasure_geometry())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let Some((data_blocks, parity_blocks)) = geometry else {
|
||||
return fallback_quorum;
|
||||
};
|
||||
let Some(write_quorum) = write_quorum_from_erasure(data_blocks, parity_blocks) else {
|
||||
return fallback_quorum;
|
||||
};
|
||||
write_quorum
|
||||
.saturating_sub(unreachable_disks)
|
||||
.max(data_blocks)
|
||||
.max(fallback_quorum)
|
||||
}
|
||||
|
||||
impl TryFrom<FileMetaVersion> for FileMetaShallowVersion {
|
||||
type Error = Error;
|
||||
|
||||
@@ -737,26 +676,6 @@ pub(super) struct V2ObjectSliceSummary {
|
||||
}
|
||||
|
||||
impl FileMetaVersion {
|
||||
fn erasure_geometry(&self) -> Option<(usize, usize)> {
|
||||
Some(match self.version_type {
|
||||
VersionType::Object | VersionType::Legacy => {
|
||||
if let Some(object) = &self.object {
|
||||
(object.erasure_m, object.erasure_n)
|
||||
} else if let Some(object) = &self.legacy_object {
|
||||
(object.erasure.data_blocks, object.erasure.parity_blocks)
|
||||
} else {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
_ => return None,
|
||||
})
|
||||
}
|
||||
|
||||
fn erasure_write_quorum(&self) -> Option<usize> {
|
||||
let (data_blocks, parity_blocks) = self.erasure_geometry()?;
|
||||
write_quorum_from_erasure(data_blocks, parity_blocks)
|
||||
}
|
||||
|
||||
/// Fast borrowed scanner for `V2Obj` slices. Returning `Ok` guarantees
|
||||
/// `FileMetaVersion::try_from` would succeed with the identical summary fields;
|
||||
/// any non-`V2Obj` or invalid input returns `Err` so callers fall back to `try_from`.
|
||||
@@ -1775,14 +1694,6 @@ impl FileMetaVersionHeader {
|
||||
self.ec_m > 0 && self.ec_n > 0
|
||||
}
|
||||
|
||||
pub fn write_quorum(&self, fallback_quorum: usize) -> usize {
|
||||
if self.version_type != VersionType::Object || !self.has_ec() {
|
||||
return fallback_quorum;
|
||||
}
|
||||
|
||||
write_quorum_from_erasure(usize::from(self.ec_m), usize::from(self.ec_n)).unwrap_or(fallback_quorum)
|
||||
}
|
||||
|
||||
pub fn matches_not_strict(&self, o: &FileMetaVersionHeader) -> bool {
|
||||
let mut ok = self.version_id == o.version_id && self.version_type == o.version_type && self.matches_ec(o);
|
||||
// Disk-loaded headers keep the null version as Some(nil), not None: all null
|
||||
@@ -4167,15 +4078,49 @@ pub fn merge_file_meta_versions(
|
||||
.iter()
|
||||
.map(|v| v.iter().map(|ver| (&ver.header, ver.meta.as_slice())).collect())
|
||||
.collect();
|
||||
merge_file_meta_borrowed_streams(quorum, strict, requested_versions, false, 0, streams)
|
||||
merge_file_meta_borrowed_streams(quorum, strict, requested_versions, streams)
|
||||
}
|
||||
|
||||
/// Picks the most common header among the tops that share `latest`'s version
|
||||
/// id; a tie keeps the header that sorts first (the newest).
|
||||
fn most_common_header_for_version_id<'a>(
|
||||
tops: &[(&'a FileMetaVersionHeader, &'a [u8])],
|
||||
mut latest: (&'a FileMetaVersionHeader, &'a [u8]),
|
||||
strict: bool,
|
||||
) -> ((&'a FileMetaVersionHeader, &'a [u8]), usize) {
|
||||
let version_id = latest.0.version_id;
|
||||
let comparable = |header: &FileMetaVersionHeader| {
|
||||
let mut header = header.clone();
|
||||
if !strict {
|
||||
header.signature = [0; 4];
|
||||
}
|
||||
header
|
||||
};
|
||||
let mut counts: HashMap<FileMetaVersionHeader, usize> = HashMap::new();
|
||||
for &top in tops {
|
||||
if top.0.version_id == version_id {
|
||||
*counts.entry(comparable(top.0)).or_insert(0) += 1;
|
||||
}
|
||||
}
|
||||
let mut latest_count = 0;
|
||||
for (header, count) in &counts {
|
||||
if *count < latest_count || (*count == latest_count && latest.0.sorts_before(header)) {
|
||||
continue;
|
||||
}
|
||||
for &top in tops {
|
||||
if comparable(top.0) == *header {
|
||||
latest = top;
|
||||
}
|
||||
}
|
||||
latest_count = *count;
|
||||
}
|
||||
(latest, latest_count)
|
||||
}
|
||||
|
||||
pub(crate) fn merge_file_meta_borrowed_streams(
|
||||
mut quorum: usize,
|
||||
mut strict: bool,
|
||||
requested_versions: usize,
|
||||
enforce_write_quorum: bool,
|
||||
write_quorum_slack: usize,
|
||||
mut streams: Vec<Vec<(&FileMetaVersionHeader, &[u8])>>,
|
||||
) -> Vec<FileMetaShallowVersion> {
|
||||
let to_shallow = |(header, meta): (&FileMetaVersionHeader, &[u8])| FileMetaShallowVersion {
|
||||
@@ -4192,32 +4137,13 @@ pub(crate) fn merge_file_meta_borrowed_streams(
|
||||
}
|
||||
|
||||
if streams.len() == 1 {
|
||||
if !enforce_write_quorum {
|
||||
return streams[0].iter().copied().map(to_shallow).collect();
|
||||
}
|
||||
|
||||
let required_quorum = streams[0]
|
||||
.first()
|
||||
.map(|&(hdr, meta)| listing_quorum_from_header_and_meta(hdr, meta, quorum, write_quorum_slack))
|
||||
.unwrap_or(quorum);
|
||||
if streams.len() >= required_quorum {
|
||||
return streams[0].iter().copied().map(to_shallow).collect();
|
||||
}
|
||||
return Vec::new();
|
||||
return streams[0].iter().copied().map(to_shallow).collect();
|
||||
}
|
||||
|
||||
if quorum == 1 {
|
||||
strict = true;
|
||||
}
|
||||
|
||||
let required_quorum = |(hdr, meta): (&FileMetaVersionHeader, &[u8])| {
|
||||
if enforce_write_quorum {
|
||||
listing_quorum_from_header_and_meta(hdr, meta, quorum, write_quorum_slack)
|
||||
} else {
|
||||
quorum
|
||||
}
|
||||
};
|
||||
|
||||
let mut tops: Vec<(&FileMetaVersionHeader, &[u8])> = Vec::with_capacity(streams.len());
|
||||
|
||||
let mut n_versions = 0;
|
||||
@@ -4249,7 +4175,7 @@ pub(crate) fn merge_file_meta_borrowed_streams(
|
||||
let latest: (&FileMetaVersionHeader, &[u8]);
|
||||
if consistent {
|
||||
latest = tops[0];
|
||||
if tops.len() >= required_quorum(latest) {
|
||||
if tops.len() >= quorum {
|
||||
merged.push(to_shallow(latest));
|
||||
if !latest.0.free_version() {
|
||||
n_versions += 1;
|
||||
@@ -4258,6 +4184,7 @@ pub(crate) fn merge_file_meta_borrowed_streams(
|
||||
} else {
|
||||
let mut latest_ref = tops[0];
|
||||
let mut latest_count = 1;
|
||||
let mut settled_by_version_id = false;
|
||||
for &ver in &tops[1..] {
|
||||
if ver.0 == latest_ref.0 {
|
||||
latest_count += 1;
|
||||
@@ -4281,39 +4208,26 @@ pub(crate) fn merge_file_meta_borrowed_streams(
|
||||
}
|
||||
|
||||
if ver.0.version_id == latest_ref.0.version_id {
|
||||
let mut x: HashMap<FileMetaVersionHeader, usize> = HashMap::new();
|
||||
for &a in &tops {
|
||||
if a.0.version_id != ver.0.version_id {
|
||||
continue;
|
||||
}
|
||||
let mut hdr = a.0.clone();
|
||||
if !strict {
|
||||
hdr.signature = [0; 4];
|
||||
}
|
||||
*x.entry(hdr).or_insert(0) += 1;
|
||||
}
|
||||
latest_count = 0;
|
||||
for (k, v) in x.iter() {
|
||||
if *v < latest_count || (*v == latest_count && latest_ref.0.sorts_before(k)) {
|
||||
continue;
|
||||
}
|
||||
for &a in &tops {
|
||||
let mut hdr = a.0.clone();
|
||||
if !strict {
|
||||
hdr.signature = [0; 4];
|
||||
}
|
||||
if hdr == *k {
|
||||
latest_ref = a;
|
||||
}
|
||||
}
|
||||
|
||||
latest_count = *v;
|
||||
}
|
||||
(latest_ref, latest_count) = most_common_header_for_version_id(&tops, latest_ref, strict);
|
||||
settled_by_version_id = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Readers return their heads in any order. When every older header
|
||||
// sharing the newest header's version id (an interrupted null-version
|
||||
// overwrite) came first, the scan above never compared them, and the
|
||||
// stream trim below would then drop that older committed generation.
|
||||
if !settled_by_version_id
|
||||
&& tops.iter().any(|ver| {
|
||||
ver.0 != latest_ref.0
|
||||
&& ver.0.version_id == latest_ref.0.version_id
|
||||
&& (strict || !ver.0.matches_not_strict(latest_ref.0))
|
||||
})
|
||||
{
|
||||
(latest_ref, latest_count) = most_common_header_for_version_id(&tops, latest_ref, strict);
|
||||
}
|
||||
latest = latest_ref;
|
||||
if latest_count >= required_quorum(latest) {
|
||||
if latest_count >= quorum {
|
||||
if !latest.0.free_version() {
|
||||
n_versions += 1;
|
||||
}
|
||||
@@ -5102,39 +5016,6 @@ mod tests {
|
||||
wr
|
||||
}
|
||||
|
||||
fn encode_legacy_v2_object_body(data_blocks: usize, parity_blocks: usize) -> Vec<u8> {
|
||||
let drive_count = data_blocks + parity_blocks;
|
||||
let payload = LegacyObjectVersionFixture {
|
||||
version_type: LegacyObjectVersionTypeFixture::Object,
|
||||
object: Some(LegacyObjectFixture {
|
||||
version_id: Some(sample_version_id().as_bytes().to_vec()),
|
||||
data_dir: Some(Uuid::from_u128(42).as_bytes().to_vec()),
|
||||
erasure_algorithm: "ReedSolomon".to_string(),
|
||||
erasure_m: data_blocks,
|
||||
erasure_n: parity_blocks,
|
||||
erasure_block_size: 1_048_576,
|
||||
erasure_index: 1,
|
||||
erasure_dist: (1..=drive_count)
|
||||
.map(|idx| u8::try_from(idx).expect("test drive index should fit u8"))
|
||||
.collect(),
|
||||
bitrot_checksum_algo: "HighwayHash".to_string(),
|
||||
part_numbers: vec![1],
|
||||
part_etags: vec!["etag-1".to_string()],
|
||||
part_sizes: vec![11],
|
||||
part_actual_sizes: vec![11],
|
||||
part_indices: vec![Vec::new()],
|
||||
size: 11,
|
||||
mod_time: Some(sample_mod_time()),
|
||||
meta_sys: HashMap::new(),
|
||||
meta_user: HashMap::from([("content-type".to_string(), "text/plain".to_string())]),
|
||||
}),
|
||||
delete_marker: None,
|
||||
write_version: 3,
|
||||
};
|
||||
|
||||
rmp_serde::to_vec_named(&payload).expect("legacy object payload should marshal")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn version_header_unmarshal_v1_uses_legacy_layout_defaults() {
|
||||
let expected = sample_header();
|
||||
@@ -5169,80 +5050,6 @@ mod tests {
|
||||
assert_eq!(decoded.ec_m, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shallow_version_write_quorum_uses_legacy_object_payload_when_header_lacks_ec() {
|
||||
let expected = sample_header();
|
||||
let encoded = encode_v2_header(&expected);
|
||||
let mut header = FileMetaVersionHeader::default();
|
||||
header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
|
||||
|
||||
let version = FileMetaShallowVersion {
|
||||
header,
|
||||
meta: encode_legacy_v2_object_body(7, 1),
|
||||
};
|
||||
|
||||
assert_eq!(version.header.write_quorum(5), 5);
|
||||
assert_eq!(version.write_quorum(5), 7);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn listing_quorum_uses_each_versions_data_floor() {
|
||||
for (data, parity, offline, fallback, expected) in
|
||||
[(8, 8, 0, 9, 9), (8, 8, 8, 8, 8), (12, 4, 8, 8, 12), (12, 4, 0, 9, 12)]
|
||||
{
|
||||
let version = FileMetaShallowVersion {
|
||||
header: FileMetaVersionHeader {
|
||||
version_type: VersionType::Object,
|
||||
ec_m: data,
|
||||
ec_n: parity,
|
||||
..Default::default()
|
||||
},
|
||||
// Modern headers must not decode the payload to decide quorum.
|
||||
meta: vec![0xff],
|
||||
};
|
||||
assert_eq!(version.listing_quorum(fallback, offline), expected);
|
||||
}
|
||||
let marker = FileMetaShallowVersion {
|
||||
header: FileMetaVersionHeader {
|
||||
version_type: VersionType::Delete,
|
||||
..Default::default()
|
||||
},
|
||||
meta: Vec::new(),
|
||||
};
|
||||
assert_eq!(marker.listing_quorum(9, 0), 9, "healthy marker needs majority metadata");
|
||||
assert_eq!(marker.listing_quorum(8, 8), 8, "degraded marker retains metadata floor");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn listing_quorum_preserves_legacy_and_zero_parity_data_floor() {
|
||||
for (data, parity, fallback, expected) in [(12, 4, 8, 12), (8, 8, 8, 8), (4, 0, 2, 4)] {
|
||||
let version = FileMetaShallowVersion {
|
||||
header: FileMetaVersionHeader {
|
||||
version_type: VersionType::Object,
|
||||
..Default::default()
|
||||
},
|
||||
meta: encode_legacy_v2_object_body(data, parity),
|
||||
};
|
||||
assert_eq!(version.listing_quorum(fallback, 8), expected);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shallow_version_write_quorum_uses_zero_parity_object_payload() {
|
||||
let expected = sample_header();
|
||||
let encoded = encode_v2_header(&expected);
|
||||
let mut header = FileMetaVersionHeader::default();
|
||||
header.unmarshal_v(2, &encoded).expect("legacy v2 header should decode");
|
||||
|
||||
let version = FileMetaShallowVersion {
|
||||
header,
|
||||
meta: encode_legacy_v2_object_body(4, 0),
|
||||
};
|
||||
|
||||
assert_eq!(version.header.write_quorum(2), 2);
|
||||
assert_eq!(version.write_quorum(2), 4);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn version_header_unmarshal_v3_round_trips_current_layout() {
|
||||
let expected = sample_header();
|
||||
|
||||
+118
-745
File diff suppressed because it is too large
Load Diff
@@ -44,7 +44,7 @@ shown. The manifest contains their complete module paths.
|
||||
| MPU delete quorum | `abort_enforces_delete_write_quorum_boundary` | Abort requires the existing delete write quorum. |
|
||||
| MPU physical publication | `complete_multipart_timeout_keeps_namespace_owner_until_physical_publication` | Timeout cannot retire an active physical publication. |
|
||||
| Restore failures | `multipart_restore_aborts_every_post_create_failure` | Every injected post-create failure aborts the remote upload. |
|
||||
| Partial LIST metadata | `rustfs-filemeta`: `resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge` | A partial newer version remains hidden. |
|
||||
| Partial LIST metadata | `rustfs-filemeta`: `resolve_merge_keeps_the_newest_version_at_listing_quorum` | A newer version below the listing quorum remains hidden behind the committed version. |
|
||||
| Metadata observations | `metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction` and its companion tests | Pending, absent, corrupt, offline and successful disk slots retain the existing quorum decisions across all four-slot arrival orders. |
|
||||
| Metadata cache bodies | `metadata_cache_real_get_and_range_preserve_body_across_read_modes` | Real cache misses and hits return identical full-body and non-zero Range bytes for verified inline/non-inline objects, with early-stop enabled/disabled and one offline disk. Hits share a complete four-slot snapshot containing all available disks and issue no new metadata reads. |
|
||||
| Late metadata identity | `late_metadata_refresh_rejects_changed_generation_and_mapped_shards` | Metadata from a real four-disk PUT still reaches selection quorum, but changed generation or incorrect per-slot shard mapping fails the disk-backed late-refresh boundary with an object-scoped read-quorum error. |
|
||||
@@ -100,10 +100,13 @@ Removing only the two old #7049 limit guards still passes because the recursive
|
||||
child's stop result provides another protection. Removing those guards and
|
||||
ignoring that result in the same stack-draining loop makes raw limit 1 return
|
||||
`a/first` and `a0` instead of only `a/first`. This tests the historical invariant,
|
||||
not a literal revert of #7049. Setting the current offline write-quorum
|
||||
allowance to zero makes the #7010 corpus return `InsufficientReadQuorum` where
|
||||
not a literal revert of #7049. At that main, setting the offline write-quorum
|
||||
allowance to zero made the #7010 corpus return `InsufficientReadQuorum` where
|
||||
the independent oracle requires visible objects. Restoring both production
|
||||
files passes both tests. These local controls do not establish a concurrent
|
||||
files passes both tests. The listing now takes its quorum as `ceil(asked/2)`
|
||||
of the online drives it asked (MinIO's rule), so the #7010 boundary needs two
|
||||
copies among three online drives; a write-quorum requirement of three would
|
||||
hide those objects again. These local controls do not establish a concurrent
|
||||
cross-page snapshot or process-crash safety.
|
||||
|
||||
## Write owner and completion routes
|
||||
|
||||
Reference in New Issue
Block a user