mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 16:48:58 +00:00
perf(ecstore): backfill rename_data old size and gate the PUT prelookup (#4598)
This commit is contained in:
Zhengchao An
@@ -261,9 +261,9 @@ pub mod data_usage {
|
||||
DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend,
|
||||
load_compression_total_from_memory, load_data_usage_from_backend, record_bucket_delete_marker_memory,
|
||||
record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
|
||||
record_compression_total_memory, refresh_bucket_usage_from_object_layer,
|
||||
refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend,
|
||||
replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
|
||||
record_bucket_object_write_unknown_previous_memory, record_compression_total_memory,
|
||||
refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
|
||||
remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -273,9 +273,9 @@ pub mod disk {
|
||||
pub use crate::disk::{
|
||||
BATCH_READ_VERSION_MAX_ITEMS, BUCKET_META_PREFIX, BatchReadVersionItem, BatchReadVersionReq, BatchReadVersionResp,
|
||||
CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
|
||||
FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp,
|
||||
ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
|
||||
validate_batch_read_version_item_count,
|
||||
FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, OldCurrentSize, RUSTFS_META_BUCKET, ReadMultipleReq,
|
||||
ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
|
||||
new_disk, validate_batch_read_version_item_count,
|
||||
};
|
||||
pub use bytes::Bytes;
|
||||
pub use endpoint::Endpoint;
|
||||
|
||||
@@ -3060,6 +3060,7 @@ mod tests {
|
||||
let response = RenameDataResp {
|
||||
old_data_dir: Some(Uuid::new_v4()),
|
||||
sign: Some(vec![1_u8; 32]),
|
||||
old_current_size: Some(crate::disk::OldCurrentSize::Present(4096)),
|
||||
};
|
||||
let json = serde_json::to_vec(&response).expect("rename data response json should encode");
|
||||
let named_msgpack = encode_msgpack_named(&response).expect("rename data response named msgpack should encode");
|
||||
|
||||
@@ -389,6 +389,22 @@ impl crate::storage_api_contracts::object::ObjectIO for Sets {
|
||||
}
|
||||
}
|
||||
|
||||
impl Sets {
|
||||
/// `put_object` plus the rename_data old-size backfill
|
||||
/// (rustfs/backlog#1009); see `SetDisks::put_object_with_old_current_size`.
|
||||
pub async fn put_object_with_old_current_size(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: &mut PutObjReader,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
|
||||
self.get_disks_by_key(object)
|
||||
.put_object_with_old_current_size(bucket, object, data, opts)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl BucketOperations for Sets {
|
||||
type Error = Error;
|
||||
|
||||
@@ -689,6 +689,34 @@ async fn record_bucket_object_write_memory_inner(
|
||||
entry.stale_snapshot_pending = false;
|
||||
}
|
||||
|
||||
/// Degraded in-memory update for an object write whose previous current size
|
||||
/// could not be determined (rustfs/backlog#1009: the pre-PUT lookup was
|
||||
/// skipped and the rename_data backfill came back unknown — mixed-version
|
||||
/// peers or sub-quorum metadata divergence). Applies only the components that
|
||||
/// are correct regardless of the previous state: the new bytes always count,
|
||||
/// and a versioned write always adds a version. objects_count (and the
|
||||
/// non-versioned overwrite's old-size subtraction) are left to the next
|
||||
/// scanner refresh, which replaces this cache with authoritative numbers.
|
||||
pub async fn record_bucket_object_write_unknown_previous_memory(bucket: &str, new_size: u64, creates_new_version: bool) {
|
||||
ensure_bucket_usage_cached(bucket).await;
|
||||
|
||||
let mut cache = memory_cache().write().await;
|
||||
let entry = cache
|
||||
.entry(bucket.to_string())
|
||||
.or_insert_with(|| cached_bucket_usage_now(BucketUsageInfo::default()));
|
||||
|
||||
entry.usage.size = entry.usage.size.saturating_add(new_size);
|
||||
if creates_new_version {
|
||||
entry.usage.versions_count = entry.usage.versions_count.saturating_add(1);
|
||||
}
|
||||
|
||||
let now = SystemTime::now();
|
||||
entry.refreshed_at = now;
|
||||
entry.usage_updated_at = now;
|
||||
entry.dirty = true;
|
||||
entry.stale_snapshot_pending = false;
|
||||
}
|
||||
|
||||
/// Fast in-memory increment for immediate quota consistency.
|
||||
pub async fn increment_bucket_usage_memory(bucket: &str, size_increment: u64) {
|
||||
record_bucket_object_write_memory(bucket, None, size_increment).await;
|
||||
@@ -1470,6 +1498,57 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: with an unknown previous state, only the
|
||||
/// always-correct components are recorded — bytes are added and a
|
||||
/// versioned write adds a version, but objects_count never moves.
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn memory_overlay_unknown_previous_adds_size_only_for_unversioned_write() {
|
||||
clear_usage_memory_cache_for_test().await;
|
||||
|
||||
let persisted = data_usage_info_for_test("bucket-a", 1, 10, SystemTime::now() - Duration::from_secs(10));
|
||||
replace_bucket_usage_memory_from_info(&persisted).await;
|
||||
record_bucket_object_write_unknown_previous_memory("bucket-a", 20, false).await;
|
||||
|
||||
let mut response = persisted.clone();
|
||||
apply_bucket_usage_memory_overlay(&mut response).await;
|
||||
|
||||
assert_eq!(response.objects_total_count, 1);
|
||||
assert_eq!(response.versions_total_count, 1);
|
||||
assert_eq!(response.objects_total_size, 30);
|
||||
assert_eq!(
|
||||
response
|
||||
.buckets_usage
|
||||
.get("bucket-a")
|
||||
.map(|usage| (usage.objects_count, usage.versions_count, usage.size)),
|
||||
Some((1, 1, 30))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn memory_overlay_unknown_previous_adds_size_and_version_for_versioned_write() {
|
||||
clear_usage_memory_cache_for_test().await;
|
||||
|
||||
let persisted = data_usage_info_for_test("bucket-a", 1, 10, SystemTime::now() - Duration::from_secs(10));
|
||||
replace_bucket_usage_memory_from_info(&persisted).await;
|
||||
record_bucket_object_write_unknown_previous_memory("bucket-a", 20, true).await;
|
||||
|
||||
let mut response = persisted.clone();
|
||||
apply_bucket_usage_memory_overlay(&mut response).await;
|
||||
|
||||
assert_eq!(response.objects_total_count, 1);
|
||||
assert_eq!(response.versions_total_count, 2);
|
||||
assert_eq!(response.objects_total_size, 30);
|
||||
assert_eq!(
|
||||
response
|
||||
.buckets_usage
|
||||
.get("bucket-a")
|
||||
.map(|usage| (usage.objects_count, usage.versions_count, usage.size)),
|
||||
Some((1, 2, 30))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn memory_overlay_records_delete_marker_without_removing_versions() {
|
||||
|
||||
@@ -18,7 +18,7 @@ use crate::disk::disk_store::get_object_disk_read_timeout;
|
||||
use crate::disk::{
|
||||
BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
|
||||
CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics,
|
||||
FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET,
|
||||
FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET,
|
||||
RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE,
|
||||
STORAGE_FORMAT_FILE_BACKUP, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int,
|
||||
endpoint::Endpoint,
|
||||
@@ -4076,6 +4076,50 @@ fn rename_data_versions_signature(meta: &FileMeta) -> Option<Vec<u8>> {
|
||||
Some(signature)
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: observe the destination key's *current* (latest)
|
||||
/// version in the dst `xl.meta` that `rename_data` already loaded, before
|
||||
/// `add_version` commits the incoming one. Replicates the pre-PUT
|
||||
/// `get_object_info` outcome bit for bit, as the per-disk `read_version`
|
||||
/// pipeline (`get_file_info`) would report it:
|
||||
///
|
||||
/// - `dst_meta_existed == false` (no dst xl.meta on this disk): the lookup
|
||||
/// errors FileNotFound → app-level `None` → `Absent`.
|
||||
/// - Existing meta whose versions are all hidden free versions: the lookup
|
||||
/// errors FileNotFound the same way → `Absent`.
|
||||
/// - Existing meta with zero versions: `get_file_info` synthesizes a deleted
|
||||
/// `FileInfo` with size 0 and the lookup returns `Ok` → `Present(0)`.
|
||||
/// - A resolvable latest version — live object, delete marker, or a
|
||||
/// purge-pending version flagged `deleted` — returns `Ok` with
|
||||
/// `ObjectInfo.size == fi.size` (0 for markers) → `Present(fi.size)`.
|
||||
/// Delete markers deliberately do NOT map to `Absent`: today's lookup
|
||||
/// returns `Ok(size 0)` for them, and delete-marker creation never
|
||||
/// decrements `objects_count`, so `Some(0)` is what keeps versioned
|
||||
/// accounting bit-identical.
|
||||
/// - A latest version that fails to decode — including the part-array length
|
||||
/// guard that `all_parts=true` enables, the same flag the per-disk lookup
|
||||
/// uses — yields `None` (unknown): the old lookup surfaced a per-disk error
|
||||
/// there, so this disk must not vote in the set-level quorum reduction.
|
||||
///
|
||||
/// One deliberate divergence: `read_data` stays `false` (the lookup used
|
||||
/// `true`), so a corrupt inline-data map that would have errored the old
|
||||
/// lookup votes the version's own `size` here instead of abstaining. The
|
||||
/// size field decodes independently of the inline data, so the vote carries
|
||||
/// the same value healthy disks report, and skipping the lookup's inline
|
||||
/// bytes clone keeps the observation allocation-light.
|
||||
fn observe_old_current_size(dst_meta_existed: bool, xlmeta: &FileMeta) -> Option<OldCurrentSize> {
|
||||
if !dst_meta_existed {
|
||||
return Some(OldCurrentSize::Absent);
|
||||
}
|
||||
if xlmeta.versions.is_empty() {
|
||||
return Some(OldCurrentSize::Present(0));
|
||||
}
|
||||
match xlmeta.into_fileinfo("", "", "", false, false, true) {
|
||||
Ok(fi) => Some(OldCurrentSize::Present(fi.size)),
|
||||
Err(rustfs_filemeta::Error::FileNotFound) => Some(OldCurrentSize::Absent),
|
||||
Err(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_root_path(path: impl AsRef<Path>) -> bool {
|
||||
path.as_ref().components().count() == 1 && path.as_ref().has_root()
|
||||
}
|
||||
@@ -5108,13 +5152,27 @@ impl DiskAPI for LocalDisk {
|
||||
};
|
||||
|
||||
let mut xlmeta = FileMeta::new();
|
||||
if let Some(dst_buf) = has_dst_buf.as_ref()
|
||||
&& FileMeta::is_xl2_v1_format(dst_buf)
|
||||
&& let Ok(nmeta) = FileMeta::load(dst_buf)
|
||||
{
|
||||
xlmeta = nmeta
|
||||
// An existing dst xl.meta that fails to parse leaves `xlmeta` empty
|
||||
// and gets overwritten by the commit below (pre-existing behavior);
|
||||
// track that so the old-size observation reports unknown instead of
|
||||
// a false `Absent` (rustfs/backlog#1009).
|
||||
let mut dst_meta_unparsable = false;
|
||||
if let Some(dst_buf) = has_dst_buf.as_ref() {
|
||||
if FileMeta::is_xl2_v1_format(dst_buf)
|
||||
&& let Ok(nmeta) = FileMeta::load(dst_buf)
|
||||
{
|
||||
xlmeta = nmeta
|
||||
} else {
|
||||
dst_meta_unparsable = true;
|
||||
}
|
||||
}
|
||||
|
||||
let old_current_size = if dst_meta_unparsable {
|
||||
None
|
||||
} else {
|
||||
observe_old_current_size(has_dst_buf.is_some(), &xlmeta)
|
||||
};
|
||||
|
||||
let mut skip_parent = dst_volume_dir.clone();
|
||||
if has_dst_buf.as_ref().is_some()
|
||||
&& let Some(parent) = dst_file_path.parent()
|
||||
@@ -5325,6 +5383,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(RenameDataResp {
|
||||
old_data_dir: has_old_data_dir,
|
||||
sign: version_signature,
|
||||
old_current_size,
|
||||
})
|
||||
} else {
|
||||
// Inline: merge read + parse + write + rename into single spawn_blocking
|
||||
@@ -5338,7 +5397,7 @@ impl DiskAPI for LocalDisk {
|
||||
None
|
||||
};
|
||||
|
||||
let (old_data_dir, version_signature) = tokio::task::spawn_blocking(move || {
|
||||
let (old_data_dir, version_signature, old_current_size) = tokio::task::spawn_blocking(move || {
|
||||
// Read existing xl.meta
|
||||
let has_dst_buf = match std::fs::read(&dst) {
|
||||
Ok(buf) => Some(Bytes::from(buf)),
|
||||
@@ -5347,13 +5406,26 @@ impl DiskAPI for LocalDisk {
|
||||
};
|
||||
|
||||
let mut xlmeta = FileMeta::new();
|
||||
if let Some(ref buf) = has_dst_buf
|
||||
&& FileMeta::is_xl2_v1_format(buf)
|
||||
&& let Ok(nmeta) = FileMeta::load(buf)
|
||||
{
|
||||
xlmeta = nmeta
|
||||
// Same as the non-inline branch: an unparsable existing dst
|
||||
// xl.meta must surface as unknown, not `Absent`
|
||||
// (rustfs/backlog#1009).
|
||||
let mut dst_meta_unparsable = false;
|
||||
if let Some(ref buf) = has_dst_buf {
|
||||
if FileMeta::is_xl2_v1_format(buf)
|
||||
&& let Ok(nmeta) = FileMeta::load(buf)
|
||||
{
|
||||
xlmeta = nmeta
|
||||
} else {
|
||||
dst_meta_unparsable = true;
|
||||
}
|
||||
}
|
||||
|
||||
let old_current_size = if dst_meta_unparsable {
|
||||
None
|
||||
} else {
|
||||
observe_old_current_size(has_dst_buf.is_some(), &xlmeta)
|
||||
};
|
||||
|
||||
let version_id = fi.version_id.unwrap_or_default();
|
||||
let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id));
|
||||
if let Some(d) = old_data_dir.as_ref() {
|
||||
@@ -5450,7 +5522,11 @@ impl DiskAPI for LocalDisk {
|
||||
}
|
||||
}
|
||||
|
||||
Ok::<(Option<uuid::Uuid>, Option<Vec<u8>>), std::io::Error>((old_data_dir, version_signature))
|
||||
Ok::<(Option<uuid::Uuid>, Option<Vec<u8>>, Option<OldCurrentSize>), std::io::Error>((
|
||||
old_data_dir,
|
||||
version_signature,
|
||||
old_current_size,
|
||||
))
|
||||
})
|
||||
.await
|
||||
.map_err(DiskError::from)??;
|
||||
@@ -5465,6 +5541,7 @@ impl DiskAPI for LocalDisk {
|
||||
Ok(RenameDataResp {
|
||||
old_data_dir,
|
||||
sign: version_signature,
|
||||
old_current_size,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -7669,6 +7746,266 @@ mod test {
|
||||
.join(STORAGE_FORMAT_FILE)
|
||||
.exists()
|
||||
);
|
||||
// rustfs/backlog#1009: the overwritten live current version (size 1
|
||||
// from `test_file_info`) must be surfaced through the backfill.
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(1)));
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: `observe_old_current_size` must mirror the pre-PUT
|
||||
/// `get_object_info` semantics bit for bit — latest version's
|
||||
/// `ObjectInfo.size` (0 for a delete-marker latest, which that lookup
|
||||
/// returns as `Ok`, not as not-found), missing key → `Absent` — and
|
||||
/// `rename_data` must report it for both the inline and non-inline commit
|
||||
/// branches.
|
||||
mod old_current_size_backfill {
|
||||
use super::*;
|
||||
use tempfile::tempdir;
|
||||
|
||||
fn live_file_info(name: &str, version_id: Uuid, size: i64, mod_time: OffsetDateTime) -> FileInfo {
|
||||
FileInfo {
|
||||
name: name.to_string(),
|
||||
version_id: Some(version_id),
|
||||
size,
|
||||
mod_time: Some(mod_time),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn delete_marker_file_info(name: &str, version_id: Uuid, mod_time: OffsetDateTime) -> FileInfo {
|
||||
FileInfo {
|
||||
name: name.to_string(),
|
||||
version_id: Some(version_id),
|
||||
deleted: true,
|
||||
mod_time: Some(mod_time),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_reports_absent_for_missing_key() {
|
||||
assert_eq!(observe_old_current_size(false, &FileMeta::default()), Some(OldCurrentSize::Absent));
|
||||
}
|
||||
|
||||
/// An existing xl.meta with zero versions reads back through
|
||||
/// `get_file_info` as a synthetic deleted FileInfo of size 0, so the
|
||||
/// pre-PUT lookup reported `Some(0)`, not not-found.
|
||||
#[test]
|
||||
fn observe_reports_present_zero_for_existing_versionless_meta() {
|
||||
assert_eq!(observe_old_current_size(true, &FileMeta::default()), Some(OldCurrentSize::Present(0)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_reports_latest_live_version_size() {
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let mut meta = FileMeta::new();
|
||||
meta.add_version(live_file_info("object", Uuid::new_v4(), 10, now - time::Duration::seconds(10)))
|
||||
.expect("older live version should be added");
|
||||
meta.add_version(live_file_info("object", Uuid::new_v4(), 42, now))
|
||||
.expect("newer live version should be added");
|
||||
|
||||
assert_eq!(observe_old_current_size(true, &meta), Some(OldCurrentSize::Present(42)));
|
||||
}
|
||||
|
||||
/// The pre-PUT lookup returns `Ok(size 0)` for a delete-marker latest
|
||||
/// (RustFS's `SetDisks::get_object_info` does not convert markers to
|
||||
/// not-found), and delete-marker creation never decrements
|
||||
/// objects_count — so the backfill must report `Present(0)` here, not
|
||||
/// `Absent`, to keep versioned accounting bit-identical.
|
||||
#[test]
|
||||
fn observe_reports_present_zero_for_delete_marker_latest() {
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let mut meta = FileMeta::new();
|
||||
meta.add_version(live_file_info("object", Uuid::new_v4(), 42, now - time::Duration::seconds(10)))
|
||||
.expect("live version should be added");
|
||||
meta.add_version(delete_marker_file_info("object", Uuid::new_v4(), now))
|
||||
.expect("delete marker should be added");
|
||||
|
||||
assert_eq!(observe_old_current_size(true, &meta), Some(OldCurrentSize::Present(0)));
|
||||
}
|
||||
|
||||
/// A latest version whose part arrays are corrupt (lengths disagree)
|
||||
/// made the old per-disk lookup error out — this disk must abstain
|
||||
/// (`None`), never vote. Pins the `all_parts=true` flag that enables
|
||||
/// the part-array length guard.
|
||||
#[test]
|
||||
fn observe_abstains_for_corrupt_part_arrays() {
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let mut fi = live_file_info("object", Uuid::new_v4(), 42, now);
|
||||
fi.add_object_part(1, "etag".to_string(), 42, Some(now), 42, None, None);
|
||||
let mut version = rustfs_filemeta::FileMetaVersion::from(fi);
|
||||
version
|
||||
.object
|
||||
.as_mut()
|
||||
.expect("object version should carry a MetaObject")
|
||||
.part_sizes
|
||||
.clear();
|
||||
|
||||
let mut meta = FileMeta::new();
|
||||
meta.add_version_filemata(version)
|
||||
.expect("corrupt-part version should still insert");
|
||||
|
||||
assert_eq!(observe_old_current_size(true, &meta), None);
|
||||
}
|
||||
|
||||
async fn test_disk(dir: &tempfile::TempDir) -> LocalDisk {
|
||||
let endpoint =
|
||||
Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
|
||||
ensure_test_volume(&disk, "bucket").await;
|
||||
ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
|
||||
disk
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inline_rename_data_reports_absent_for_fresh_key() {
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let disk = test_disk(&dir).await;
|
||||
|
||||
let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-fresh", new_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("inline rename_data should commit");
|
||||
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Absent));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inline_rename_data_reports_present_zero_for_delete_marker_latest() {
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let disk = test_disk(&dir).await;
|
||||
|
||||
let now = OffsetDateTime::now_utc();
|
||||
let mut old_meta = FileMeta::new();
|
||||
old_meta
|
||||
.add_version(live_file_info("object", Uuid::new_v4(), 42, now - time::Duration::seconds(10)))
|
||||
.expect("live version should be added");
|
||||
old_meta
|
||||
.add_version(delete_marker_file_info("object", Uuid::new_v4(), now))
|
||||
.expect("delete marker should be added");
|
||||
let dst_object_dir = dir.path().join("bucket").join("object");
|
||||
fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
|
||||
fs::write(
|
||||
dst_object_dir.join(STORAGE_FORMAT_FILE),
|
||||
old_meta.marshal_msg().expect("old metadata should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("old metadata should be written");
|
||||
|
||||
let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-marker", new_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("inline rename_data should commit");
|
||||
|
||||
// The pre-PUT lookup returns Ok(size 0) for a marker latest, so
|
||||
// the backfill must match it (see observe_old_current_size docs).
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(0)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inline_rename_data_reports_unknown_for_unparsable_dst_meta() {
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let disk = test_disk(&dir).await;
|
||||
|
||||
let dst_object_dir = dir.path().join("bucket").join("object");
|
||||
fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
|
||||
fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), b"not-an-xl-meta")
|
||||
.await
|
||||
.expect("garbage metadata should be written");
|
||||
|
||||
let new_fi = test_file_info("object", Uuid::new_v4(), None, Some(Bytes::from_static(b"inline-new")));
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-garbage", new_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("inline rename_data should commit");
|
||||
|
||||
assert_eq!(resp.old_current_size, None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn non_inline_rename_data_reports_absent_then_previous_size() {
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let disk = test_disk(&dir).await;
|
||||
|
||||
// First non-inline commit: fresh key must report Absent.
|
||||
let first_data_dir = Uuid::new_v4();
|
||||
let tmp_data_dir = dir
|
||||
.path()
|
||||
.join(RUSTFS_META_TMP_BUCKET)
|
||||
.join("tmp-first")
|
||||
.join(first_data_dir.to_string());
|
||||
fs::create_dir_all(&tmp_data_dir)
|
||||
.await
|
||||
.expect("tmp data dir should be created");
|
||||
fs::write(tmp_data_dir.join("part.1"), b"first")
|
||||
.await
|
||||
.expect("part should be written");
|
||||
let mut first_fi = test_file_info("object", Uuid::new_v4(), Some(first_data_dir), None);
|
||||
first_fi.size = 5;
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-first", first_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("first non-inline rename_data should commit");
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Absent));
|
||||
|
||||
// Overwrite: the committed live version (size 5) must be reported.
|
||||
let second_data_dir = Uuid::new_v4();
|
||||
let tmp_data_dir = dir
|
||||
.path()
|
||||
.join(RUSTFS_META_TMP_BUCKET)
|
||||
.join("tmp-second")
|
||||
.join(second_data_dir.to_string());
|
||||
fs::create_dir_all(&tmp_data_dir)
|
||||
.await
|
||||
.expect("tmp data dir should be created");
|
||||
fs::write(tmp_data_dir.join("part.1"), b"second-longer")
|
||||
.await
|
||||
.expect("part should be written");
|
||||
let mut second_fi = test_file_info("object", Uuid::new_v4(), Some(second_data_dir), None);
|
||||
second_fi.size = 13;
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-second", second_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("second non-inline rename_data should commit");
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(5)));
|
||||
}
|
||||
|
||||
/// Twin of the inline unparsable-dst test: the `dst_meta_unparsable`
|
||||
/// tracking is duplicated per branch, so the non-inline copy needs its
|
||||
/// own regression coverage.
|
||||
#[tokio::test]
|
||||
async fn non_inline_rename_data_reports_unknown_for_unparsable_dst_meta() {
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let disk = test_disk(&dir).await;
|
||||
|
||||
let dst_object_dir = dir.path().join("bucket").join("object");
|
||||
fs::create_dir_all(&dst_object_dir).await.expect("dst dir should be created");
|
||||
fs::write(dst_object_dir.join(STORAGE_FORMAT_FILE), b"not-an-xl-meta")
|
||||
.await
|
||||
.expect("garbage metadata should be written");
|
||||
|
||||
let data_dir = Uuid::new_v4();
|
||||
let tmp_data_dir = dir
|
||||
.path()
|
||||
.join(RUSTFS_META_TMP_BUCKET)
|
||||
.join("tmp-garbage-noninline")
|
||||
.join(data_dir.to_string());
|
||||
fs::create_dir_all(&tmp_data_dir)
|
||||
.await
|
||||
.expect("tmp data dir should be created");
|
||||
fs::write(tmp_data_dir.join("part.1"), b"payload")
|
||||
.await
|
||||
.expect("part should be written");
|
||||
let mut new_fi = test_file_info("object", Uuid::new_v4(), Some(data_dir), None);
|
||||
new_fi.size = 7;
|
||||
let resp = disk
|
||||
.rename_data(RUSTFS_META_TMP_BUCKET, "tmp-garbage-noninline", new_fi, "bucket", "object")
|
||||
.await
|
||||
.expect("non-inline rename_data should commit over garbage metadata");
|
||||
|
||||
assert_eq!(resp.old_current_size, None);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -839,10 +839,43 @@ pub struct DiskOption {
|
||||
pub health_check: bool,
|
||||
}
|
||||
|
||||
/// Per-disk observation of the destination key's *current* (latest) version
|
||||
/// in the dst `xl.meta` that `rename_data` reads before it commits the
|
||||
/// incoming version (rustfs/backlog#1009). Mirrors the pre-PUT
|
||||
/// `get_object_info` outcome — which the app layer previously obtained with an
|
||||
/// extra full-disk fanout — bit for bit: `Absent` iff that lookup would report
|
||||
/// object-not-found (missing xl.meta, or only hidden free versions), and
|
||||
/// `Present(size)` iff it would return `Ok` with `ObjectInfo.size == size`.
|
||||
/// Note a delete-marker latest is `Present(0)`, not `Absent`: the lookup
|
||||
/// returns markers as `Ok(size 0)` and delete-marker accounting never
|
||||
/// decrements objects_count, so `Present(0)` is what keeps usage numbers
|
||||
/// identical.
|
||||
///
|
||||
/// The parity target is the set-level (`SetDisks`) lookup. Two app-visible
|
||||
/// lookup variants deviated from it and from each other — the multi-pool
|
||||
/// `ECStore` path converts a delete-marker latest to not-found, and
|
||||
/// directory-key lookups pin the nil version — and both deviations
|
||||
/// over-incremented objects_count relative to delete-marker accounting, so
|
||||
/// the backfill standardizes on the self-consistent set-level answer.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
|
||||
pub enum OldCurrentSize {
|
||||
/// The pre-PUT lookup would have reported object-not-found for this key.
|
||||
Absent,
|
||||
/// The pre-PUT lookup would have returned `Ok` with this `ObjectInfo.size`
|
||||
/// (0 for a delete-marker latest).
|
||||
Present(i64),
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Serialize, Deserialize)]
|
||||
pub struct RenameDataResp {
|
||||
pub old_data_dir: Option<Uuid>,
|
||||
pub sign: Option<Vec<u8>>,
|
||||
/// `None` means unknown — the disk could not determine the previous
|
||||
/// current version (pre-#1009 peer on the wire, or an existing dst
|
||||
/// `xl.meta` that failed to parse). Consumers must treat unknown as
|
||||
/// "cannot vote", never as `Absent`.
|
||||
#[serde(default)]
|
||||
pub old_current_size: Option<OldCurrentSize>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
@@ -1229,10 +1262,56 @@ mod tests {
|
||||
let resp = RenameDataResp {
|
||||
old_data_dir: Some(uuid),
|
||||
sign: Some(signature.clone()),
|
||||
old_current_size: Some(OldCurrentSize::Present(42)),
|
||||
};
|
||||
|
||||
assert_eq!(resp.old_data_dir, Some(uuid));
|
||||
assert_eq!(resp.sign, Some(signature));
|
||||
assert_eq!(resp.old_current_size, Some(OldCurrentSize::Present(42)));
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: `old_current_size` must survive a named-msgpack
|
||||
/// round trip (the internode RPC encoding) for every variant.
|
||||
#[test]
|
||||
fn test_rename_data_resp_old_current_size_msgpack_roundtrip() {
|
||||
for old_current_size in [None, Some(OldCurrentSize::Absent), Some(OldCurrentSize::Present(1337))] {
|
||||
let resp = RenameDataResp {
|
||||
old_data_dir: Some(Uuid::new_v4()),
|
||||
sign: Some(vec![0x01, 0x02, 0x03]),
|
||||
old_current_size,
|
||||
};
|
||||
|
||||
let encoded = rmp_serde::encode::to_vec_named(&resp).expect("named msgpack should encode");
|
||||
let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("named msgpack should decode");
|
||||
|
||||
assert_eq!(decoded.old_data_dir, resp.old_data_dir);
|
||||
assert_eq!(decoded.sign, resp.sign);
|
||||
assert_eq!(decoded.old_current_size, resp.old_current_size);
|
||||
}
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: a payload from a peer that predates
|
||||
/// `old_current_size` must decode with the field defaulting to `None`
|
||||
/// (unknown), keeping mixed-version clusters wire-compatible.
|
||||
#[test]
|
||||
fn test_rename_data_resp_decodes_payload_without_old_current_size() {
|
||||
#[derive(Serialize)]
|
||||
struct LegacyRenameDataResp {
|
||||
old_data_dir: Option<Uuid>,
|
||||
sign: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
let legacy = LegacyRenameDataResp {
|
||||
old_data_dir: Some(Uuid::new_v4()),
|
||||
sign: Some(vec![0x0a, 0x0b]),
|
||||
};
|
||||
|
||||
let encoded = rmp_serde::encode::to_vec_named(&legacy).expect("legacy named msgpack should encode");
|
||||
let decoded: RenameDataResp = rmp_serde::decode::from_slice(&encoded).expect("legacy payload should decode");
|
||||
|
||||
assert_eq!(decoded.old_data_dir, legacy.old_data_dir);
|
||||
assert_eq!(decoded.sign, legacy.sign);
|
||||
assert_eq!(decoded.old_current_size, None);
|
||||
}
|
||||
|
||||
/// Test constants
|
||||
|
||||
@@ -492,3 +492,179 @@ async fn blackbox_issue3031_diag_covers_put_success_cleanup_and_error_summary()
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: the `put_object` old-size backfill must reproduce, for
|
||||
/// every PUT shape, exactly what the app layer's pre-PUT `get_object_info`
|
||||
/// lookup would have observed — that is the invariant that lets the app skip
|
||||
/// the lookup without changing a single usage-accounting number.
|
||||
mod old_current_size_backfill {
|
||||
use super::*;
|
||||
use crate::disk::OldCurrentSize;
|
||||
use crate::error::is_err_object_not_found;
|
||||
use crate::set_disk::SetDisks;
|
||||
|
||||
/// What the pre-PUT lookup would report right now for `object`.
|
||||
async fn prelookup_expectation(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> OldCurrentSize {
|
||||
let opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
};
|
||||
match set_disks.get_object_info(bucket, object, &opts).await {
|
||||
Ok(object_info) => OldCurrentSize::Present(object_info.size),
|
||||
Err(err) => {
|
||||
assert!(is_err_object_not_found(&err), "prelookup expectation hit unexpected error: {err:?}");
|
||||
OldCurrentSize::Absent
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn put_and_backfill(
|
||||
set_disks: &Arc<SetDisks>,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
len: usize,
|
||||
opts: &ObjectOptions,
|
||||
) -> Option<OldCurrentSize> {
|
||||
let payload = (0..len).map(|idx| ((idx * 31) % 251) as u8).collect::<Vec<_>>();
|
||||
let mut reader = PutObjReader::from_vec(payload);
|
||||
let (_object_info, backfill) = set_disks
|
||||
.put_object_with_old_current_size(bucket, object, &mut reader, opts)
|
||||
.await
|
||||
.expect("put_object should succeed");
|
||||
backfill
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn backfill_matches_prelookup_for_every_put_shape() {
|
||||
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
|
||||
let bucket = "bb-old-current-size";
|
||||
set_disks
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("bucket should be created");
|
||||
let opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// Fresh key (inline-sized): both the lookup and the backfill say Absent.
|
||||
let object = "object.bin";
|
||||
assert_eq!(prelookup_expectation(&set_disks, bucket, object).await, OldCurrentSize::Absent);
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
|
||||
Some(OldCurrentSize::Absent)
|
||||
);
|
||||
|
||||
// Unversioned inline overwrite: the previous live size must surface.
|
||||
let expected = prelookup_expectation(&set_disks, bucket, object).await;
|
||||
assert_eq!(expected, OldCurrentSize::Present(1024));
|
||||
assert_eq!(put_and_backfill(&set_disks, bucket, object, 2048, &opts).await, Some(expected));
|
||||
|
||||
// Non-inline overwrite (payload above the inline block threshold).
|
||||
let expected = prelookup_expectation(&set_disks, bucket, object).await;
|
||||
assert_eq!(expected, OldCurrentSize::Present(2048));
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, object, BLOCK_SIZE_V2 + 123, &opts).await,
|
||||
Some(expected)
|
||||
);
|
||||
|
||||
// Overwriting a non-inline object reports its size back.
|
||||
let expected = prelookup_expectation(&set_disks, bucket, object).await;
|
||||
assert_eq!(expected, OldCurrentSize::Present((BLOCK_SIZE_V2 + 123) as i64));
|
||||
assert_eq!(put_and_backfill(&set_disks, bucket, object, 64, &opts).await, Some(expected));
|
||||
|
||||
// Versioned writes: a new version over a live latest reports the
|
||||
// latest's size, not the incoming version's.
|
||||
let versioned = "versioned.bin";
|
||||
let first_version_opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
versioned: true,
|
||||
version_id: Some(uuid::Uuid::new_v4().to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, versioned, 111, &first_version_opts).await,
|
||||
Some(OldCurrentSize::Absent)
|
||||
);
|
||||
let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
|
||||
assert_eq!(expected, OldCurrentSize::Present(111));
|
||||
let second_version_opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
versioned: true,
|
||||
version_id: Some(uuid::Uuid::new_v4().to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, versioned, 222, &second_version_opts).await,
|
||||
Some(expected)
|
||||
);
|
||||
|
||||
// Delete-marker latest: today's lookup returns the marker as
|
||||
// Ok(size 0) — NOT not-found — and delete-marker creation never
|
||||
// decremented objects_count, so the backfill must reproduce
|
||||
// Present(0) to keep versioned accounting identical.
|
||||
set_disks
|
||||
.delete_object(
|
||||
bucket,
|
||||
versioned,
|
||||
ObjectOptions {
|
||||
no_lock: true,
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("versioned delete should create a delete marker");
|
||||
let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
|
||||
assert_eq!(expected, OldCurrentSize::Present(0));
|
||||
let third_version_opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
versioned: true,
|
||||
version_id: Some(uuid::Uuid::new_v4().to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, versioned, 333, &third_version_opts).await,
|
||||
Some(expected)
|
||||
);
|
||||
}
|
||||
|
||||
/// Sub-quorum divergence through the real set-level fanout: when only 2 of
|
||||
/// 4 disks can still decode the destination's xl.meta (write quorum is 3),
|
||||
/// the PUT must still commit but the backfill must come back unknown —
|
||||
/// never a fabricated `Absent`/`Present` from a below-quorum vote.
|
||||
#[tokio::test]
|
||||
async fn backfill_is_unknown_below_quorum_but_put_succeeds() {
|
||||
let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
|
||||
let bucket = "bb-old-current-subquorum";
|
||||
set_disks
|
||||
.make_bucket(bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("bucket should be created");
|
||||
let opts = ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let object = "object.bin";
|
||||
assert_eq!(
|
||||
put_and_backfill(&set_disks, bucket, object, 1024, &opts).await,
|
||||
Some(OldCurrentSize::Absent)
|
||||
);
|
||||
|
||||
// Corrupt the committed xl.meta on two disks: those observations turn
|
||||
// unknown, leaving 2 definite votes < write quorum 3.
|
||||
{
|
||||
let disks = set_disks.disks.read().await;
|
||||
for disk in disks.iter().flatten().take(2) {
|
||||
let meta_path = disk.path().join(bucket).join(object).join("xl.meta");
|
||||
fs::write(&meta_path, b"not-an-xl-meta")
|
||||
.await
|
||||
.expect("xl.meta should be overwritable with garbage");
|
||||
}
|
||||
}
|
||||
|
||||
let backfill = put_and_backfill(&set_disks, bucket, object, 2048, &opts).await;
|
||||
assert_eq!(backfill, None, "a below-quorum vote must surface as unknown");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,6 +46,7 @@ use crate::diagnostics::get::{
|
||||
GetObjectFailureReason, classify_disk_error, get_stage_timer_if_enabled, record_get_object_pipeline_failure,
|
||||
record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
|
||||
};
|
||||
use crate::disk::OldCurrentSize;
|
||||
use crate::erasure::coding::BitrotReader;
|
||||
use crate::io_support::bitrot::{
|
||||
BitrotReaderStageMetrics, DeferredReaderStripeHandle, create_bitrot_reader_with_stage_metrics,
|
||||
@@ -2490,7 +2491,13 @@ impl SetDisks {
|
||||
dst_bucket: &str,
|
||||
dst_object: &str,
|
||||
write_quorum: usize,
|
||||
) -> disk::error::Result<(Vec<Option<DiskStore>>, Option<Vec<u8>>, Option<Uuid>, Vec<Option<DiskStore>>)> {
|
||||
) -> disk::error::Result<(
|
||||
Vec<Option<DiskStore>>,
|
||||
Option<Vec<u8>>,
|
||||
Option<Uuid>,
|
||||
Vec<Option<DiskStore>>,
|
||||
Option<OldCurrentSize>,
|
||||
)> {
|
||||
let mut futures = Vec::with_capacity(disks.len());
|
||||
|
||||
let mut errs = Vec::with_capacity(disks.len());
|
||||
@@ -2528,6 +2535,7 @@ impl SetDisks {
|
||||
|
||||
let mut disk_versions = vec![None; disks.len()];
|
||||
let mut data_dirs = vec![None; disks.len()];
|
||||
let mut old_current_sizes = vec![None; disks.len()];
|
||||
|
||||
let results = join_all(futures).await;
|
||||
|
||||
@@ -2536,6 +2544,7 @@ impl SetDisks {
|
||||
Ok(res) => {
|
||||
data_dirs[idx] = res.old_data_dir;
|
||||
disk_versions[idx].clone_from(&res.sign);
|
||||
old_current_sizes[idx] = res.old_current_size;
|
||||
errs.push(None);
|
||||
}
|
||||
Err(e) => {
|
||||
@@ -2639,6 +2648,7 @@ impl SetDisks {
|
||||
|
||||
let data_dir = Self::reduce_common_data_dir(&data_dirs, write_quorum);
|
||||
let versions = Self::select_rename_data_versions(&disk_versions, &errs, write_quorum);
|
||||
let old_current_size = Self::reduce_common_old_current_size(&old_current_sizes, write_quorum);
|
||||
let online_disks = Self::eval_disks(disks, &errs);
|
||||
let cleanup_disks = if let Some(data_dir) = data_dir {
|
||||
disks
|
||||
@@ -2657,7 +2667,35 @@ impl SetDisks {
|
||||
vec![None; disks.len()]
|
||||
};
|
||||
|
||||
Ok((online_disks, versions, data_dir, cleanup_disks))
|
||||
Ok((online_disks, versions, data_dir, cleanup_disks, old_current_size))
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: reduce the per-disk observations of the
|
||||
/// destination's previous current version to one set-level value, mirroring
|
||||
/// `reduce_common_data_dir`: the observation reported by at least
|
||||
/// `write_quorum` disks wins; anything short of that (disk errors, unknown
|
||||
/// votes from pre-#1009 peers or unparsable metadata, genuine divergence)
|
||||
/// yields `None` (unknown). Unknown per-disk entries never vote.
|
||||
pub(in crate::set_disk) fn reduce_common_old_current_size(
|
||||
old_current_sizes: &[Option<OldCurrentSize>],
|
||||
write_quorum: usize,
|
||||
) -> Option<OldCurrentSize> {
|
||||
let mut counts: HashMap<OldCurrentSize, usize> = HashMap::new();
|
||||
|
||||
for observation in old_current_sizes.iter().flatten().copied() {
|
||||
*counts.entry(observation).or_insert(0) += 1;
|
||||
}
|
||||
|
||||
let mut max = 0;
|
||||
let mut old_current_size = None;
|
||||
for (observation, count) in counts {
|
||||
if count > max {
|
||||
max = count;
|
||||
old_current_size = Some(observation);
|
||||
}
|
||||
}
|
||||
|
||||
if max >= write_quorum { old_current_size } else { None }
|
||||
}
|
||||
|
||||
pub(in crate::set_disk) fn reduce_common_versions(disk_versions: &[Option<Vec<u8>>], write_quorum: usize) -> Option<Vec<u8>> {
|
||||
@@ -4664,6 +4702,76 @@ mod tests {
|
||||
assert!(!r.has_residue());
|
||||
}
|
||||
|
||||
// rustfs/backlog#1009: quorum reduction of the per-disk old-current-size
|
||||
// observations returned by rename_data.
|
||||
mod reduce_common_old_current_size {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn agreement_at_quorum_wins() {
|
||||
let observations = vec![
|
||||
Some(OldCurrentSize::Present(5)),
|
||||
Some(OldCurrentSize::Present(5)),
|
||||
Some(OldCurrentSize::Present(5)),
|
||||
Some(OldCurrentSize::Absent),
|
||||
];
|
||||
assert_eq!(
|
||||
SetDisks::reduce_common_old_current_size(&observations, 3),
|
||||
Some(OldCurrentSize::Present(5))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn absent_is_a_definite_vote() {
|
||||
let observations = vec![
|
||||
Some(OldCurrentSize::Absent),
|
||||
Some(OldCurrentSize::Absent),
|
||||
Some(OldCurrentSize::Absent),
|
||||
None,
|
||||
];
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), Some(OldCurrentSize::Absent));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn divergence_below_quorum_is_unknown() {
|
||||
let observations = vec![
|
||||
Some(OldCurrentSize::Present(5)),
|
||||
Some(OldCurrentSize::Present(7)),
|
||||
Some(OldCurrentSize::Absent),
|
||||
Some(OldCurrentSize::Absent),
|
||||
];
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_disks_do_not_vote() {
|
||||
// Two agreeing disks plus two unknowns must not fabricate quorum.
|
||||
let observations = vec![Some(OldCurrentSize::Present(5)), Some(OldCurrentSize::Present(5)), None, None];
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
|
||||
}
|
||||
|
||||
/// Kills the "unknown counts as an Absent vote" mutant: a full set of
|
||||
/// unknowns (rolling upgrade, every peer pre-#1009) must stay unknown —
|
||||
/// fabricating `Absent` would record "new object" on every overwrite
|
||||
/// and inflate objects_count.
|
||||
#[test]
|
||||
fn all_unknown_is_unknown() {
|
||||
let observations: Vec<Option<OldCurrentSize>> = vec![None; 4];
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_majority_does_not_become_absent_quorum() {
|
||||
let observations = vec![Some(OldCurrentSize::Present(5)), None, None, None];
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&observations, 3), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_observations_are_unknown() {
|
||||
assert_eq!(SetDisks::reduce_common_old_current_size(&[], 2), None);
|
||||
}
|
||||
}
|
||||
|
||||
// A2: not-found normalized to success (parity with MinIO commitRenameDataDir).
|
||||
#[test]
|
||||
fn cleanup_receipt_not_found_counts_as_reclaimed() {
|
||||
|
||||
@@ -1420,7 +1420,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
|
||||
let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
|
||||
|
||||
let (online_disks, versions, op_old_dir, cleanup_disks) = Self::rename_data(
|
||||
// The trailing `_` drops the rename_data old-size backfill
|
||||
// (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
|
||||
// `get_object_info` lookup, so the backfill has no consumer here yet.
|
||||
let (online_disks, versions, op_old_dir, cleanup_disks, _) = Self::rename_data(
|
||||
&shuffle_disks,
|
||||
RUSTFS_META_MULTIPART_BUCKET,
|
||||
&upload_id_path,
|
||||
|
||||
@@ -21,6 +21,8 @@
|
||||
|
||||
use super::super::*;
|
||||
|
||||
use crate::disk::OldCurrentSize;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
type Error = Error;
|
||||
@@ -556,8 +558,29 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
Ok(reader)
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self, data,))]
|
||||
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
|
||||
self.put_object_with_old_current_size(bucket, object, data, opts)
|
||||
.await
|
||||
.map(|(object_info, _)| object_info)
|
||||
}
|
||||
}
|
||||
|
||||
impl SetDisks {
|
||||
/// `put_object` plus the destination key's previous current-version size,
|
||||
/// quorum-reduced from the dst `xl.meta` copies `rename_data` reads while
|
||||
/// committing (rustfs/backlog#1009). `None` means unknown (mixed-version
|
||||
/// peers, unparsable metadata, or sub-quorum divergence) — callers must
|
||||
/// fall back to degraded accounting, never assume "absent". The extra
|
||||
/// value is deliberately *not* part of `ObjectInfo`, which feeds S3
|
||||
/// responses, event payloads, replication, and ILM verbatim.
|
||||
#[tracing::instrument(skip(self, data,))]
|
||||
pub async fn put_object_with_old_current_size(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: &mut PutObjReader,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
|
||||
crate::hp_guard!("SetDisks::put_object");
|
||||
self.invalidate_get_object_metadata_cache(bucket, object).await;
|
||||
|
||||
@@ -629,7 +652,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
|
||||
let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
|
||||
|
||||
let result: Result<ObjectInfo> = async {
|
||||
let result: Result<(ObjectInfo, Option<OldCurrentSize>)> = async {
|
||||
let erasure = coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
|
||||
let put_object_size = known_put_object_storage_size(data.size());
|
||||
@@ -888,7 +911,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
}
|
||||
|
||||
let rename_stage_start = Instant::now();
|
||||
let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data(
|
||||
let (online_disks, _, op_old_dir, cleanup_disks, old_current_size) = Self::rename_data(
|
||||
&shuffle_disks,
|
||||
RUSTFS_META_TMP_BUCKET,
|
||||
tmp_dir.as_str(),
|
||||
@@ -1023,7 +1046,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
|
||||
);
|
||||
}
|
||||
|
||||
Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
|
||||
Ok((
|
||||
ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended),
|
||||
old_current_size,
|
||||
))
|
||||
}
|
||||
.await;
|
||||
|
||||
|
||||
@@ -386,9 +386,31 @@ impl crate::storage_api_contracts::object::ObjectIO for ECStore {
|
||||
}
|
||||
#[instrument(level = "debug", skip(self, data))]
|
||||
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
|
||||
let result =
|
||||
enqueue_transition_after_write(self.handle_put_object(bucket, object, data, opts).await, LcEventSrc::S3PutObject)
|
||||
.await;
|
||||
self.put_object_with_old_current_size(bucket, object, data, opts)
|
||||
.await
|
||||
.map(|(object_info, _)| object_info)
|
||||
}
|
||||
}
|
||||
|
||||
impl ECStore {
|
||||
/// `put_object` plus the rename_data old-size backfill
|
||||
/// (rustfs/backlog#1009); see `SetDisks::put_object_with_old_current_size`.
|
||||
/// Post-write hooks (immediate ILM transition enqueue, list-cache
|
||||
/// invalidation) match the plain `put_object` path exactly.
|
||||
#[instrument(level = "debug", skip(self, data))]
|
||||
pub async fn put_object_with_old_current_size(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
data: &mut PutObjReader,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(ObjectInfo, Option<crate::disk::OldCurrentSize>)> {
|
||||
let result = match self.handle_put_object(bucket, object, data, opts).await {
|
||||
Ok((object_info, old_current_size)) => enqueue_transition_after_write(Ok(object_info), LcEventSrc::S3PutObject)
|
||||
.await
|
||||
.map(|object_info| (object_info, old_current_size)),
|
||||
Err(err) => Err(err),
|
||||
};
|
||||
if result.is_ok() {
|
||||
list_objects::observe_list_objects_mutation(self, bucket).await;
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use super::*;
|
||||
use crate::disk::OldCurrentSize;
|
||||
use crate::set_disk::{
|
||||
get_lock_acquire_timeout, get_object_lock_diag_slow_acquire_threshold, get_object_lock_diag_slow_hold_threshold,
|
||||
is_lock_optimization_enabled, is_object_lock_diag_enabled,
|
||||
@@ -718,7 +719,7 @@ impl ECStore {
|
||||
object: &str,
|
||||
data: &mut PutObjReader,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<ObjectInfo> {
|
||||
) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
|
||||
check_put_object_args(bucket, object)?;
|
||||
|
||||
let object = encode_dir_object(object);
|
||||
@@ -726,7 +727,9 @@ impl ECStore {
|
||||
// Keep PUT atomic-read friendly: SetDisks takes the object write lock only
|
||||
// around precondition checks and the final rename/commit.
|
||||
if self.single_pool() {
|
||||
return self.pools[0].put_object(bucket, object.as_str(), data, opts).await;
|
||||
return self.pools[0]
|
||||
.put_object_with_old_current_size(bucket, object.as_str(), data, opts)
|
||||
.await;
|
||||
}
|
||||
|
||||
let idx = if opts.data_movement && opts.version_id.is_some() {
|
||||
@@ -744,7 +747,9 @@ impl ECStore {
|
||||
));
|
||||
}
|
||||
|
||||
self.pools[idx].put_object(bucket, &object, data, opts).await
|
||||
self.pools[idx]
|
||||
.put_object_with_old_current_size(bucket, &object, data, opts)
|
||||
.await
|
||||
}
|
||||
|
||||
#[instrument(skip(self))]
|
||||
|
||||
@@ -32,3 +32,5 @@ mod capacity_dirty_scope_test;
|
||||
mod delete_objects_stat_gating_test;
|
||||
#[cfg(test)]
|
||||
mod lifecycle_transition_api_test;
|
||||
#[cfg(test)]
|
||||
mod put_prelookup_gating_test;
|
||||
|
||||
@@ -19,7 +19,6 @@ use rustfs_io_metrics::buffered_write;
|
||||
|
||||
use crate::storage_api::table::get_bucket_metadata;
|
||||
|
||||
use super::storage_api::object_usecase::ECStore;
|
||||
use super::storage_api::object_usecase::access::{
|
||||
PostObjectRequestMarker, authorize_request, has_bypass_governance_header, req_info_mut,
|
||||
};
|
||||
@@ -64,7 +63,7 @@ use super::storage_api::object_usecase::contract::object::{ObjectIO as _, Object
|
||||
use super::storage_api::object_usecase::contract::range::HTTPRangeSpec;
|
||||
use super::storage_api::object_usecase::data_usage::{
|
||||
record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
|
||||
record_bucket_object_write_memory,
|
||||
record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory,
|
||||
};
|
||||
use super::storage_api::object_usecase::deadlock_detector;
|
||||
use super::storage_api::object_usecase::ecfs::FS;
|
||||
@@ -93,6 +92,7 @@ use super::storage_api::object_usecase::sse::{
|
||||
};
|
||||
use super::storage_api::object_usecase::storage_class as storageclass;
|
||||
use super::storage_api::object_usecase::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper};
|
||||
use super::storage_api::object_usecase::{ECStore, OldCurrentSize};
|
||||
use super::storage_api::object_usecase::{
|
||||
RFC1123, check_preconditions, get_validated_store, has_replication_rules, parse_object_lock_legal_hold,
|
||||
parse_object_lock_retention, parse_part_number_i32_to_usize, remove_object_lock_metadata_for_copy,
|
||||
@@ -3510,26 +3510,42 @@ impl DefaultObjectUsecase {
|
||||
)
|
||||
.await?;
|
||||
|
||||
let current_opts: ObjectOptions = internal_object_info_lookup_opts(
|
||||
get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
|
||||
.await
|
||||
.map_err(ApiError::from)?,
|
||||
);
|
||||
let previous_current_info = {
|
||||
crate::hp_guard!("S3::put_object_prelookup");
|
||||
store.get_object_info(&bucket, &key, ¤t_opts).await
|
||||
};
|
||||
let previous_current_size = match previous_current_info {
|
||||
Ok(existing_obj_info) => {
|
||||
validate_existing_object_lock_for_write(&existing_obj_info, &opts)?;
|
||||
Some(existing_obj_info.size.max(0) as u64)
|
||||
}
|
||||
Err(err) => {
|
||||
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
|
||||
return Err(ApiError::from(err).into());
|
||||
// rustfs/backlog#1009: the pre-PUT lookup has exactly two consumers —
|
||||
// the existing-object WORM validation and usage accounting's
|
||||
// previous_current_size. When the bucket has no object locking (WORM is
|
||||
// a provable no-op; the gate fails closed on metadata errors) and the
|
||||
// PUT targets the latest version (no explicit version_id from internal
|
||||
// replication), the lookup is skipped and accounting is backfilled from
|
||||
// the dst xl.meta that rename_data already reads, saving a full-disk
|
||||
// metadata fanout per PUT.
|
||||
let prelookup_required = version_id.is_some() || put_prelookup_worm_gate(&bucket).await;
|
||||
// Outer None = prelookup skipped (accounting comes from the commit
|
||||
// backfill); Some(inner) = the previous current size as observed by the
|
||||
// lookup, with the pre-#1009 semantics kept bit-for-bit.
|
||||
let prelookup_previous_current_size: Option<Option<u64>> = if prelookup_required {
|
||||
let current_opts: ObjectOptions = internal_object_info_lookup_opts(
|
||||
get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
|
||||
.await
|
||||
.map_err(ApiError::from)?,
|
||||
);
|
||||
let previous_current_info = {
|
||||
crate::hp_guard!("S3::put_object_prelookup");
|
||||
store.get_object_info(&bucket, &key, ¤t_opts).await
|
||||
};
|
||||
Some(match previous_current_info {
|
||||
Ok(existing_obj_info) => {
|
||||
validate_existing_object_lock_for_write(&existing_obj_info, &opts)?;
|
||||
Some(existing_obj_info.size.max(0) as u64)
|
||||
}
|
||||
None
|
||||
}
|
||||
Err(err) => {
|
||||
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
|
||||
return Err(ApiError::from(err).into());
|
||||
}
|
||||
None
|
||||
}
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let actual_size = size;
|
||||
@@ -3713,8 +3729,8 @@ impl DefaultObjectUsecase {
|
||||
}
|
||||
});
|
||||
|
||||
let obj_info = match store
|
||||
.put_object(&bucket, &key, &mut reader, &opts)
|
||||
let (obj_info, backfilled_old_current_size) = match store
|
||||
.put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
|
||||
.await
|
||||
.map_err(ApiError::from)
|
||||
{
|
||||
@@ -3764,11 +3780,35 @@ impl DefaultObjectUsecase {
|
||||
let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await;
|
||||
|
||||
let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
|
||||
// Fast in-memory update for immediate quota and admin usage consistency
|
||||
if put_versioned {
|
||||
record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
|
||||
} else {
|
||||
record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
|
||||
// Fast in-memory update for immediate quota and admin usage consistency.
|
||||
// The previous current size comes from the prelookup when it ran,
|
||||
// otherwise from the rename_data backfill (rustfs/backlog#1009); the
|
||||
// backfill reproduces the lookup's observation bit for bit (latest
|
||||
// version's ObjectInfo.size — 0 for a delete-marker latest — or
|
||||
// not-found → None).
|
||||
match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) {
|
||||
Some(previous_current_size) => {
|
||||
if put_versioned {
|
||||
record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
|
||||
} else {
|
||||
record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
|
||||
}
|
||||
}
|
||||
None => {
|
||||
// Neither source could determine the previous state (peers
|
||||
// predating the backfill field during a rolling upgrade, or
|
||||
// sub-quorum metadata divergence). Record the components that
|
||||
// are correct regardless; the next authoritative scanner
|
||||
// refresh replaces the in-memory numbers.
|
||||
debug!(
|
||||
target: "rustfs::app::object_usecase",
|
||||
bucket = %bucket,
|
||||
key = %key,
|
||||
put_versioned,
|
||||
"put_object old-size backfill unknown; recording degraded usage delta"
|
||||
);
|
||||
record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned).await;
|
||||
}
|
||||
}
|
||||
|
||||
let raw_version = obj_info.version_id.map(|v| v.to_string());
|
||||
@@ -6567,6 +6607,28 @@ async fn bucket_object_locking_enabled(bucket: &str) -> bool {
|
||||
.is_ok_and(|metadata| metadata.object_locking())
|
||||
}
|
||||
|
||||
/// Fail-closed WORM gate for skipping the pre-PUT lookup
|
||||
/// (rustfs/backlog#1009): a bucket-metadata read failure counts as "locking
|
||||
/// enabled" so the existing-object lock validation can never silently
|
||||
/// disappear on a degraded metadata subsystem.
|
||||
pub(super) async fn put_prelookup_worm_gate(bucket: &str) -> bool {
|
||||
match get_bucket_metadata(bucket).await {
|
||||
Ok(metadata) => metadata.object_locking(),
|
||||
Err(_) => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: map the rename_data old-size backfill onto the
|
||||
/// `previous_current_size` value the usage-accounting helpers expect. Outer
|
||||
/// `None` = unknown (no quorum agreement, or a peer predates the field) — the
|
||||
/// caller must fall back to the degraded accounting path.
|
||||
fn previous_current_size_from_backfill(backfill: Option<OldCurrentSize>) -> Option<Option<u64>> {
|
||||
backfill.map(|observation| match observation {
|
||||
OldCurrentSize::Present(size) => Some(size.max(0) as u64),
|
||||
OldCurrentSize::Absent => None,
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -6662,6 +6724,19 @@ mod tests {
|
||||
object_info_with_lock_metadata(metadata)
|
||||
}
|
||||
|
||||
/// rustfs/backlog#1009: the backfill→accounting mapping must mirror the
|
||||
/// prelookup exactly — a live latest version maps to `Some(size)` (clamped
|
||||
/// at 0 like the prelookup's `.max(0)`), absent/delete-marker maps to
|
||||
/// `None`, and an unknown backfill maps to outer `None` so the caller
|
||||
/// takes the degraded path instead of fabricating "new object".
|
||||
#[test]
|
||||
fn previous_current_size_from_backfill_mirrors_prelookup_semantics() {
|
||||
assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Present(42))), Some(Some(42)));
|
||||
assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Present(-7))), Some(Some(0)));
|
||||
assert_eq!(previous_current_size_from_backfill(Some(OldCurrentSize::Absent)), Some(None));
|
||||
assert_eq!(previous_current_size_from_backfill(None), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn validate_existing_object_lock_allows_versioned_new_version_with_compliance_retention() {
|
||||
let opts = ObjectOptions {
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Regression coverage for rustfs/backlog#1009: the PUT path skips its
|
||||
//! pre-PUT `get_object_info` only when `put_prelookup_worm_gate` proves the
|
||||
//! existing-object WORM validation is a no-op. These tests pin the gate's
|
||||
//! truth table against a real 4-disk `ECStore` with the bucket metadata sys
|
||||
//! initialized, mirroring the HP-8 delete-gating fixture:
|
||||
//!
|
||||
//! - a bucket created with Object Lock keeps the prelookup (gate = true);
|
||||
//! - a plain bucket takes the skip path (gate = false);
|
||||
//! - an unknown bucket (metadata lookup error) fails closed (gate = true),
|
||||
//! so a degraded metadata subsystem can never silently drop the WORM check.
|
||||
|
||||
use super::object_usecase::put_prelookup_worm_gate;
|
||||
use super::storage_api::test::bucket::metadata_sys;
|
||||
use super::storage_api::test::contract::bucket::{BucketOperations, BucketOptions, MakeBucketOptions};
|
||||
use super::storage_api::test::{ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints};
|
||||
use serial_test::serial;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use tempfile::TempDir;
|
||||
use tokio::fs;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use uuid::Uuid;
|
||||
|
||||
static PUT_GATING_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>, TempDir)> = OnceLock::new();
|
||||
|
||||
async fn setup_put_gating_env() -> Arc<ECStore> {
|
||||
if let Some((_paths, store, _)) = PUT_GATING_ENV.get() {
|
||||
return store.clone();
|
||||
}
|
||||
|
||||
let temp_dir = TempDir::new().expect("create temp dir for put prelookup gating test");
|
||||
let temp_path = temp_dir.path().to_path_buf();
|
||||
|
||||
let disk_paths = vec![
|
||||
temp_path.join("disk1"),
|
||||
temp_path.join("disk2"),
|
||||
temp_path.join("disk3"),
|
||||
temp_path.join("disk4"),
|
||||
];
|
||||
for disk_path in &disk_paths {
|
||||
fs::create_dir_all(disk_path).await.unwrap();
|
||||
}
|
||||
|
||||
let mut endpoints = Vec::new();
|
||||
for (i, disk_path) in disk_paths.iter().enumerate() {
|
||||
let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
|
||||
endpoint.set_pool_index(0);
|
||||
endpoint.set_set_index(0);
|
||||
endpoint.set_disk_index(i);
|
||||
endpoints.push(endpoint);
|
||||
}
|
||||
|
||||
let pool_endpoints = PoolEndpoints {
|
||||
legacy: false,
|
||||
set_count: 1,
|
||||
drives_per_set: 4,
|
||||
endpoints: Endpoints::from(endpoints),
|
||||
cmd_line: "put-prelookup-gating-test".to_string(),
|
||||
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
|
||||
};
|
||||
|
||||
let endpoint_pools = EndpointServerPools(vec![pool_endpoints]);
|
||||
super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
|
||||
let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let buckets_list = ecstore
|
||||
.list_bucket(&BucketOptions {
|
||||
no_metadata: true,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let buckets = buckets_list.into_iter().map(|v| v.name).collect();
|
||||
metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await;
|
||||
|
||||
let _ = PUT_GATING_ENV.set((disk_paths, ecstore.clone(), temp_dir));
|
||||
ecstore
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn worm_gate_keeps_prelookup_for_object_lock_bucket() {
|
||||
let ecstore = setup_put_gating_env().await;
|
||||
let bucket = format!("put-gate-lock-{}", Uuid::new_v4());
|
||||
|
||||
ecstore
|
||||
.make_bucket(
|
||||
&bucket,
|
||||
&MakeBucketOptions {
|
||||
lock_enabled: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create object-lock bucket");
|
||||
|
||||
assert!(
|
||||
put_prelookup_worm_gate(&bucket).await,
|
||||
"an object-lock bucket must keep the pre-PUT lookup"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn worm_gate_allows_skip_for_plain_bucket() {
|
||||
let ecstore = setup_put_gating_env().await;
|
||||
let bucket = format!("put-gate-plain-{}", Uuid::new_v4());
|
||||
|
||||
ecstore
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create plain bucket");
|
||||
|
||||
assert!(
|
||||
!put_prelookup_worm_gate(&bucket).await,
|
||||
"a bucket without object locking must take the prelookup-skip path"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn worm_gate_fails_closed_when_bucket_metadata_is_unavailable() {
|
||||
let _ecstore = setup_put_gating_env().await;
|
||||
let missing_bucket = format!("put-gate-missing-{}", Uuid::new_v4());
|
||||
|
||||
assert!(
|
||||
put_prelookup_worm_gate(&missing_bucket).await,
|
||||
"a bucket-metadata lookup failure must fail closed and keep the pre-PUT lookup"
|
||||
);
|
||||
}
|
||||
@@ -117,6 +117,19 @@ pub(crate) mod data_usage {
|
||||
.await;
|
||||
}
|
||||
|
||||
pub(crate) async fn record_bucket_object_write_unknown_previous_memory(
|
||||
bucket: &str,
|
||||
new_size: u64,
|
||||
creates_new_version: bool,
|
||||
) {
|
||||
crate::storage::storage_api::ecstore_data_usage::record_bucket_object_write_unknown_previous_memory(
|
||||
bucket,
|
||||
new_size,
|
||||
creates_new_version,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
pub(crate) async fn remove_bucket_usage_from_backend(
|
||||
store: Arc<crate::storage::storage_api::ECStore>,
|
||||
bucket: &str,
|
||||
@@ -982,9 +995,9 @@ pub(crate) mod object_usecase {
|
||||
object_utils, options, request_context, s3_api, set_disk, sse, storage_class, timeout_wrapper,
|
||||
};
|
||||
pub(crate) use crate::storage::storage_api::{
|
||||
ECStore, RFC1123, StorageDeletedObject, StorageObjectInfo, StorageObjectLockDeleteOptions, StorageObjectOptions,
|
||||
StorageObjectToDelete, StoragePutObjReader, check_preconditions, get_validated_store, has_replication_rules,
|
||||
parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize,
|
||||
ECStore, OldCurrentSize, RFC1123, StorageDeletedObject, StorageObjectInfo, StorageObjectLockDeleteOptions,
|
||||
StorageObjectOptions, StorageObjectToDelete, StoragePutObjReader, check_preconditions, get_validated_store,
|
||||
has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize,
|
||||
remove_object_lock_metadata_for_copy, strip_managed_encryption_metadata, validate_bucket_object_lock_enabled,
|
||||
validate_object_key, validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
|
||||
wrap_response_with_cors,
|
||||
|
||||
@@ -360,9 +360,9 @@ pub(crate) mod ecstore_data_usage {
|
||||
pub(crate) use rustfs_ecstore::api::data_usage::{
|
||||
apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend,
|
||||
record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
|
||||
record_bucket_object_write_memory, refresh_bucket_usage_from_object_layer,
|
||||
refresh_versioned_bucket_usage_from_object_layer, remove_bucket_usage_from_backend,
|
||||
replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
|
||||
record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory,
|
||||
refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
|
||||
remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -370,8 +370,8 @@ pub(crate) mod ecstore_data_usage {
|
||||
pub(crate) mod ecstore_disk {
|
||||
pub(crate) use rustfs_ecstore::api::disk::{
|
||||
BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore,
|
||||
FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
|
||||
RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout,
|
||||
FileInfoVersions, FileReader, FileWriter, OldCurrentSize, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp,
|
||||
ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout,
|
||||
validate_batch_read_version_item_count,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
|
||||
@@ -531,6 +531,7 @@ pub(crate) type BatchReadVersionResp = ecstore_disk::BatchReadVersionResp;
|
||||
pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq;
|
||||
pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp;
|
||||
pub(crate) type ReadOptions = ecstore_disk::ReadOptions;
|
||||
pub(crate) type OldCurrentSize = ecstore_disk::OldCurrentSize;
|
||||
pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp;
|
||||
pub(crate) type ReplicationStatusType = ecstore_bucket::replication::ReplicationStatusType;
|
||||
pub(crate) type ReplicationStats = StorageReplicationStatsHandle;
|
||||
|
||||
Reference in New Issue
Block a user