From 706ffdc9acdaa3fd1aeb758d43b346170e1f0c99 Mon Sep 17 00:00:00 2001 From: Chris Date: Thu, 17 Sep 2026 08:57:49 +0800 Subject: [PATCH] fix(ecstore): keep the single-block bound on every inline candidate (#7954) --- crates/ecstore/src/set_disk/ops/object.rs | 63 +++++++++++++++++++++-- crates/heal/src/heal/outcome.rs | 42 ++++++++++++--- 2 files changed, 96 insertions(+), 9 deletions(-) diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index c5daef759..ddf9dfe38 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3636,13 +3636,22 @@ impl SetDisks { // Transformed streams (unknown stored size) are admitted by their // plaintext size and then take the streaming encode with inline // buffer writers, since the single-block fast path needs a known length. - // Protected inline writes retain the single-block bound on proof metadata. + // Every inline candidate keeps the single-block bound on its admission + // size: the inline buffer writer grows without a cap, so an explicit + // INLINE_BLOCK budget above `block_size` must not admit multi-block + // payloads into memory and xl.meta. Protected inline writes also need + // a known stored length for their proof metadata, so they never fall + // back to the plaintext size. + let inline_admission_size = if put_object_size >= 0 || protect_write { + put_object_size + } else { + data.actual_size() + }; let is_inline_buffer = storage_class_config.should_inline( inline_admission_shard_size(&erasure, put_object_size, data.actual_size()), erasure.data_shards, opts.versioned, - ) && (!protect_write - || usize::try_from(put_object_size).is_ok_and(|size| size <= erasure.block_size)); + ) && usize::try_from(inline_admission_size).is_ok_and(|size| size <= erasure.block_size); let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled(); let shard_file_size = shard_file_size_raw; @@ -11795,6 +11804,54 @@ mod inline_put_commit_path_tests { } } + #[tokio::test] + #[serial] + async fn unprotected_puts_keep_multi_block_inline_candidates_external() { + for transformed in [true, false] { + let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let storage_class = + temp_env::with_var(INLINE_BLOCK_ENV, Some("16MiB"), || lookup_config_for_pools(&KVS::new(), &[4])) + .expect("large inline budget should resolve"); + set_disks.set_test_storage_class_config(storage_class); + let bucket = "unprotected-inline-candidates"; + let object = "object.bin"; + make_bucket(&disk_stores, bucket).await; + // Above the single erasure block: an explicit INLINE_BLOCK budget must + // not pull a multi-block payload into the unbounded inline buffer. + let plaintext = compressible_payload(1024 * 1024 + 17); + let (mut reader, opts) = if transformed { + transformed_put(plaintext.clone()) + } else { + (PutObjReader::from_vec(plaintext.clone()), ObjectOptions::default()) + }; + temp_env::async_with_vars( + [( + crate::set_disk::core::io_primitives::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, + Some("false"), + )], + set_disks.put_object(bucket, object, &mut reader, &opts), + ) + .await + .expect("multi-block PUT should commit with part files"); + + for (disk_index, disk) in disk_stores.iter().enumerate() { + let file_info = disk + .read_version("", bucket, object, "", &ReadOptions::default()) + .await + .unwrap_or_else(|err| panic!("disk {disk_index} should persist metadata: {err}")); + assert!( + !file_info.inline_data(), + "disk {disk_index}: a payload above block_size must stay external (transformed={transformed})" + ); + } + assert!( + count_part_files(&temp_dirs) > 0, + "a multi-block object must keep part files (transformed={transformed})" + ); + assert_eq!(read_back(&set_disks, bucket, object).await, plaintext); + } + } + #[tokio::test] async fn transformed_put_above_inline_budget_keeps_part_files() { let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index 640d8ed6f..619ef3ea7 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -105,13 +105,17 @@ pub struct HealObjectReceipt { impl HealObjectReceipt { pub(crate) fn verified_for(&self, expected: &HealObjectIdentity) -> bool { - matches!( - self.disposition, + let disposition_verifies = match self.disposition { HealObjectDisposition::Repaired - | HealObjectDisposition::VerifiedHealthy - | HealObjectDisposition::MetadataHealthy - | HealObjectDisposition::AuthoritativelyAbsent - ) && self.identity.kind == expected.kind + | HealObjectDisposition::VerifiedHealthy + | HealObjectDisposition::AuthoritativelyAbsent => true, + // A metadata/presence proof never certifies payload bytes, so it + // cannot discharge a request that exists to decode the payload. + HealObjectDisposition::MetadataHealthy => expected.kind != HealObjectKind::Decode, + _ => false, + }; + disposition_verifies + && self.identity.kind == expected.kind && self.identity.bucket == expected.bucket && self.identity.object == expected.object && self.identity.version_id == expected.version_id @@ -530,6 +534,32 @@ mod canonical_outcome_tests { assert!(serde_json::from_value::(value).is_err()); } + #[test] + fn metadata_health_receipt_never_discharges_a_payload_decode_request() { + let incarnation = Uuid::new_v4(); + let mut expected = HealObjectIdentity { + bucket_incarnation_id: Some(incarnation), + ..item(HealObjectDisposition::Unknown).identity + }; + let mut receipt = HealObjectReceipt { + identity: expected.clone(), + disposition: HealObjectDisposition::MetadataHealthy, + }; + assert!( + receipt.verified_for(&expected), + "a presence proof still settles a metadata-level object request" + ); + + expected.kind = HealObjectKind::Decode; + receipt.identity.kind = HealObjectKind::Decode; + assert!( + !receipt.verified_for(&expected), + "a presence-only proof must not clear a payload decode responsibility" + ); + receipt.disposition = HealObjectDisposition::VerifiedHealthy; + assert!(receipt.verified_for(&expected)); + } + #[test] fn canonical_outcome_categories_have_one_terminal_count() { let mut outcome = HealTaskOutcome::default();