fix(s3): short-circuit conditional reads at metadata quorum (#8459)

* fix(s3): evaluate conditional reads before opening object bodies

Reuse locked metadata snapshots for conditional GETs and unify HEAD
preconditions and cache headers. Preserve body-cache hooks and validate
encryption metadata and key authorization before returning bodyless errors.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(s3): unify conditional reads at metadata quorum

Apply scoped metadata conditions to the existing reader pipeline instead
of maintaining parallel prepared-read branches. Preserve quorum checks,
early-stop, coalescing, and cache-hook behavior on condition misses.

Track metadata-only termination through selection and reject body access
or a changed decision. Keep concrete reader futures off request stacks.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(s3): strengthen conditional read acceptance evidence

Assert zero shard, decode, cache-fill, and remote-body calls on matching
conditions with live positive controls. Exercise GET and HEAD through
s3s and Hyper with conditional-header and precedence matrices.

Add reproducible paired application-read benchmarks with isolated
metadata controls and strict drift, artifact, and resource gates.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(s3): remove obsolete conditional comparison helper

Keep ETag comparison tests on the canonical read-condition evaluator,
remove unused date-trait imports, and name the HTTP case type.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(s3): validate full conditional etag fields

Reject adjacent tags and malformed trailing list members before accepting a
match. Keep the existing request-scoped metadata read and authorization order,
while removing redundant validation wrappers and unused benchmark overrides.
Replace simulated phase coverage with real 304/412 cache-fill probes and extend
HTTP/1.1 and HTTP/2 ingress regressions.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(ecstore): cache terminal conditional metadata safely

Preserve quorum-validated conditional metadata when early termination
cancels the remaining fanout. Restrict those entries to terminal GET
readers; body and full-layout consumers still obtain their own metadata.

Serialize cache publication per key so conditional entries cannot replace
complete layouts or renew their TTL. Retain generation fencing and the
final selected-metadata predicate check before any body access.

Cover real four-disk hedging, nonterminal body and range reads, consumer
isolation, generation races, atomic publication and actual cache expiry.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(log-analyzer): align heal writer failure rule with current logs

Match the current structured all-targets failure, retain legacy log support, and reject partial or unrelated failures.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-12 16:59:21 +08:00
committed by GitHub
co-authored by heihutu zhi22915
parent 0018bbf5d7
commit dd90bb7488
32 changed files with 3641 additions and 332 deletions
+5 -3
View File
@@ -521,13 +521,14 @@ pub mod integrity {
pub mod object {
pub use crate::object_api::{
BLOCK_SIZE_V2, ERASURE_ALGORITHM, EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook,
GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReader, NamespaceLockFence, ObjectEncryptionResolver,
ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission,
RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
GetObjectBodyCacheHookLookup, GetObjectBodySource, GetObjectReadCondition, GetObjectReadMetadata, GetObjectReader,
NamespaceLockFence, ObjectEncryptionResolver, ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions,
PutObjReader, QuotaAdmission, RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitStartError,
ScannerPublicationCommitState, ShardIntegrityWriteMode, StreamConsumer, WriteCommitGuard, WriteCompletion,
get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
with_get_object_read_condition,
};
pub use crate::store::{
PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError,
@@ -592,6 +593,7 @@ pub mod set_disk {
#[cfg(feature = "test-util")]
pub mod test_util {
pub use crate::bucket::quota::reservation::fail_next_quota_ledger_save_for_test;
pub use crate::set_disk::ConditionalReadBenchmarkMetadataGuard;
pub use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause};
/// Keep a namespace commit pending until the returned owner is dropped.
+1
View File
@@ -182,6 +182,7 @@ pub(crate) const GET_METADATA_CACHE_REASON_DELETE_MARKER: &str = "delete_marker"
pub(crate) const GET_METADATA_CACHE_REASON_DIST_ERASURE: &str = "dist_erasure";
pub(crate) const GET_METADATA_CACHE_REASON_INCL_FREE_VERSIONS: &str = "incl_free_versions";
pub(crate) const GET_METADATA_CACHE_REASON_INSUFFICIENT_CACHED_QUORUM: &str = "insufficient_cached_quorum";
pub(crate) const GET_METADATA_CACHE_REASON_BODY_LAYOUT_REQUIRED: &str = "body_layout_required";
pub(crate) const GET_METADATA_CACHE_REASON_META_BUCKET: &str = "meta_bucket";
pub(crate) const GET_METADATA_CACHE_REASON_NO_LOCK: &str = "no_lock";
pub(crate) const GET_METADATA_CACHE_REASON_NOT_FOUND_OR_EXPIRED: &str = "not_found_or_expired";
+4
View File
@@ -174,10 +174,14 @@ async fn open_reader_source(
metrics_path: Option<&'static str>,
) -> disk::error::Result<Option<BoxedObjectReader>> {
if let Some(data) = inline_data {
#[cfg(test)]
crate::set_disk::disk_call_counters::record_shard_open(path, true);
let mut reader = Cursor::new(data);
reader.set_position(u64::try_from(offset).map_err(|_| DiskError::FileCorrupt)?);
Ok(Some(ShardReader::InMemory(reader)))
} else if let Some(disk) = disk {
#[cfg(test)]
crate::set_disk::disk_call_counters::record_shard_open(path, false);
open_disk_reader(disk, bucket, path, offset, length, use_mmap_read, metrics_path)
.await
.map(Some)
+4
View File
@@ -102,10 +102,14 @@ mod body_cache_hook;
mod encryption;
mod hook_slot;
mod object_mutation_hook;
mod read_condition;
mod readers;
mod types;
pub(crate) mod write_commit_context;
pub use read_condition::{GetObjectReadCondition, GetObjectReadMetadata, with_get_object_read_condition};
pub(crate) use read_condition::{get_object_read_condition_is_active, get_object_read_condition_is_terminal};
#[cfg(test)]
mod persisted_metadata_keys_tests;
@@ -0,0 +1,113 @@
// 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.
use rustfs_filemeta::{FileInfo, ObjectPartInfo};
use std::future::Future;
use std::sync::Arc;
use time::OffsetDateTime;
/// Borrowed fields needed to decide whether an object read needs a body.
pub struct GetObjectReadMetadata<'a> {
pub bucket: &'a str,
pub object: &'a str,
pub etag: Option<&'a str>,
pub mod_time: Option<OffsetDateTime>,
pub parts: &'a [ObjectPartInfo],
pub delete_marker: bool,
}
impl<'a> GetObjectReadMetadata<'a> {
pub(crate) fn from_file_info(bucket: &'a str, object: &'a str, info: &'a FileInfo) -> Self {
Self {
bucket,
object,
etag: info.metadata.get("etag").map(String::as_str),
mod_time: info.mod_time,
parts: &info.parts,
delete_marker: info.deleted,
}
}
}
/// A pure decision on quorum-validated metadata. Authorization belongs to the caller.
pub trait GetObjectReadCondition: Send + Sync {
fn is_terminal(&self, metadata: GetObjectReadMetadata<'_>) -> bool;
}
tokio::task_local! {
static READ_CONDITION: Arc<dyn GetObjectReadCondition>;
}
/// Applies a condition to this reader call, without inheriting it into shared producers.
pub async fn with_get_object_read_condition<F: Future>(condition: Arc<dyn GetObjectReadCondition>, read: F) -> F::Output {
READ_CONDITION.scope(condition, read).await
}
pub(crate) fn get_object_read_condition_is_terminal(metadata: GetObjectReadMetadata<'_>) -> bool {
READ_CONDITION
.try_with(|condition| condition.is_terminal(metadata))
.unwrap_or(false)
}
pub(crate) fn get_object_read_condition_is_active() -> bool {
READ_CONDITION.try_with(|_| ()).is_ok()
}
#[cfg(test)]
mod tests {
use super::*;
struct Stop;
impl GetObjectReadCondition for Stop {
fn is_terminal(&self, _: GetObjectReadMetadata<'_>) -> bool {
true
}
}
fn metadata() -> GetObjectReadMetadata<'static> {
GetObjectReadMetadata {
bucket: "bucket",
object: "object",
etag: Some("etag"),
mod_time: None,
parts: &[],
delete_marker: false,
}
}
#[tokio::test]
async fn read_condition_scope_does_not_escape_or_enter_shared_producers() {
assert!(!get_object_read_condition_is_terminal(metadata()));
with_get_object_read_condition(Arc::new(Stop), async {
assert!(get_object_read_condition_is_terminal(metadata()));
assert!(
!tokio::spawn(async { get_object_read_condition_is_terminal(metadata()) })
.await
.expect("shared producer")
);
})
.await;
assert!(!get_object_read_condition_is_terminal(metadata()));
}
#[tokio::test]
async fn dropping_a_pending_read_removes_its_condition_scope() {
let mut read = Box::pin(with_get_object_read_condition(Arc::new(Stop), async {
assert!(get_object_read_condition_is_terminal(metadata()));
std::future::pending::<()>().await;
}));
assert!(futures::poll!(&mut read).is_pending());
drop(read);
assert!(!get_object_read_condition_is_terminal(metadata()));
}
}
+19
View File
@@ -528,6 +528,14 @@ pub struct GetObjectReader {
}
impl GetObjectReader {
pub(crate) fn condition_stopped(object_info: ObjectInfo) -> Self {
Self {
stream: Box::new(ConditionStoppedBody),
object_info,
buffered_body: None,
body_source: GetObjectBodySource::Unprobed,
}
}
/// Builds a fully materialized reader from a cache-coordinated body.
pub fn from_cache_body(mut object_info: ObjectInfo, body: Bytes) -> Result<Self> {
object_info.size = i64::try_from(body.len()).map_err(|_| Error::other("cached GET body length exceeds i64::MAX"))?;
@@ -552,6 +560,17 @@ impl GetObjectReader {
}
}
struct ConditionStoppedBody;
impl AsyncRead for ConditionStoppedBody {
fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
if buf.remaining() == 0 {
return Poll::Ready(Ok(()));
}
Poll::Ready(Err(std::io::Error::other("object read condition must be handled before body access")))
}
}
enum ReadTransform {
// Written but never read by production code: the enclosing struct already
// carries the same pair as `storage_offset`/`storage_length`. They survive
@@ -0,0 +1,63 @@
// 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.
//! Test-util-only controls for the paired read benchmark's dedicated bucket.
use std::sync::atomic::{AtomicBool, Ordering};
static ACTIVE: AtomicBool = AtomicBool::new(false);
/// Keeps metadata off the cache in the benchmark's isolated bucket so GET
/// metadata reads exercise the configured slow disk. Body-cache policy
/// remains intact.
#[must_use]
pub struct ConditionalReadBenchmarkMetadataGuard {
_private: (),
}
impl ConditionalReadBenchmarkMetadataGuard {
pub fn acquire() -> Self {
assert!(
ACTIVE
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
);
Self { _private: () }
}
}
impl Drop for ConditionalReadBenchmarkMetadataGuard {
fn drop(&mut self) {
ACTIVE.store(false, Ordering::Release);
}
}
pub(super) fn cache_bypass_applies(bucket: &str) -> bool {
bucket == "conditional-bench" && ACTIVE.load(Ordering::Acquire)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn benchmark_metadata_profile_is_scoped_to_its_bucket_and_owner() {
assert!(!cache_bypass_applies("conditional-bench"));
let owner = ConditionalReadBenchmarkMetadataGuard::acquire();
assert!(cache_bypass_applies("conditional-bench"));
assert!(!cache_bypass_applies("other-bucket"));
drop(owner);
assert!(!cache_bypass_applies("conditional-bench"));
}
}
@@ -1260,6 +1260,10 @@ async fn try_create_bitrot_readers_via_batch_pread(
}
let requests: Vec<_> = batch_items.iter().map(|(_, p, off, len)| (p.clone(), *off, *len)).collect();
#[cfg(test)]
for (disk_index, _, _, _) in &batch_items {
disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, *disk_index);
}
let batch_results = batch_shard_pread(requests).await;
let mut setup = BitrotReaderSetup::new(disks.len());
@@ -6078,6 +6082,10 @@ pub(crate) mod disk_call_counters {
pub const KIND_READ_VERSION: &str = "read_version";
pub const KIND_BATCH_READ_VERSION: &str = "batch_read_version";
pub const KIND_METADATA_SLOWTAIL_FAULT: &str = "metadata_slowtail_fault";
pub const KIND_INLINE_SHARD_OPEN: &str = "inline_shard_open";
pub const KIND_DISK_SHARD_OPEN: &str = "disk_shard_open";
/// Body verification, direct shard assembly, and erasure decoder entry points.
pub const KIND_BODY_DECODE: &str = "body_decode";
/// Registry key: (object, kind, disk_index).
type CountKey = (String, String, usize);
@@ -6108,6 +6116,28 @@ pub(crate) mod disk_call_counters {
.or_insert(0) += 1;
}
/// Matches shard paths to the observed object's namespace path. This also
/// records deferred opens running in spawned decoder tasks.
pub(crate) fn record_shard_open(path: &str, inline: bool) {
let mut reg = registry().lock().expect("disk call-counter registry poisoned");
let objects: Vec<_> = reg
.observed
.iter()
.filter(|object| {
path == object.as_str() || path.strip_prefix(object.as_str()).is_some_and(|rest| rest.starts_with('/'))
})
.cloned()
.collect();
let kind = if inline {
KIND_INLINE_SHARD_OPEN
} else {
KIND_DISK_SHARD_OPEN
};
for object in objects {
*reg.counts.entry((object, kind.to_owned(), 0)).or_insert(0) += 1;
}
}
/// RAII scope that observes call counts for a single `object`. Counts recorded
/// while the scope is alive — including those recorded inside spawned tasks —
/// are queryable; the scope clears its own counts on drop.
@@ -66,6 +66,7 @@ use tokio::task::JoinSet;
pub(in crate::set_disk) struct MetadataReadResult {
pub(in crate::set_disk) slots: Vec<MetadataDiskObservation>,
pub(in crate::set_disk) diagnostics: MetadataFanoutDiagnostics,
pub(in crate::set_disk) condition_stopped: bool,
}
impl MetadataReadResult {
@@ -655,6 +656,8 @@ async fn inline_body_miss_reason_from_observations<'a>(
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
};
#[cfg(test)]
crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
Some(body) if body.len() == object_size => None,
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
@@ -1078,7 +1081,11 @@ impl SetDisks {
0,
);
}
Ok(MetadataReadResult { slots, diagnostics })
Ok(MetadataReadResult {
slots,
diagnostics,
condition_stopped: false,
})
}
#[allow(clippy::too_many_arguments)]
@@ -1121,6 +1128,7 @@ impl SetDisks {
let mut next_fanout_index = 0usize;
let mut scheduled_count = 0usize;
let mut force_full_wait = false;
let read_condition_active = read_data && crate::object_api::get_object_read_condition_is_active();
let mut final_miss_reason_override = None;
let mut non_inline_candidate_eligible = None;
let mut single_pending_hedge_deadline = None;
@@ -1258,12 +1266,28 @@ impl SetDisks {
}
}
if !force_full_wait
if (!force_full_wait || read_condition_active)
&& let Some(decision) = accumulator
.early_stop_decision()
.or_else(|| accumulator.version_early_stop_decision())
{
let should_return_early = if read_data {
let condition_stopped = read_condition_active
&& decision.reason == crate::diagnostics::get::GET_METADATA_EARLY_STOP_REASON_VALID_QUORUM
&& accumulator.candidate.as_ref().is_some_and(|candidate| {
!candidate.deleted
&& crate::object_api::get_object_read_condition_is_terminal(
crate::object_api::GetObjectReadMetadata::from_file_info(
bucket.as_ref(),
object.as_ref(),
candidate,
),
)
});
let should_return_early = if condition_stopped {
true
} else if force_full_wait {
false
} else if read_data {
match accumulator.candidate.as_ref() {
Some(_candidate) if non_inline_candidate_eligible == Some(true) => {
accumulator.candidate_has_read_reserve()
@@ -1332,7 +1356,11 @@ impl SetDisks {
cancelled_count,
);
let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
return Ok(MetadataReadResult { slots, diagnostics });
return Ok(MetadataReadResult {
slots,
diagnostics,
condition_stopped,
});
}
}
@@ -1410,7 +1438,11 @@ impl SetDisks {
rustfs_io_metrics::record_get_object_metadata_early_stop_saved_responses(metrics_path, 0);
rustfs_io_metrics::record_get_object_metadata_fanout_lifecycle(metrics_path, scheduled_count, scheduled_count, 0);
let diagnostics = MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations);
Ok(MetadataReadResult { slots, diagnostics })
Ok(MetadataReadResult {
slots,
diagnostics,
condition_stopped: false,
})
}
/// Test-only seam that records one per-disk `read_version` metadata RPC for
+261
View File
@@ -793,6 +793,11 @@ const DEFAULT_RUSTFS_GET_METADATA_TWO_PHASE_READ_PLAN_ENABLE: bool = false;
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
#[cfg(feature = "test-util")]
mod conditional_read_bench;
#[cfg(feature = "test-util")]
pub use conditional_read_bench::ConditionalReadBenchmarkMetadataGuard;
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
@@ -919,6 +924,7 @@ pub use ops::heal_walk::HealWalkVersion;
pub(in crate::set_disk) struct GetObjectFileInfo {
owned: Option<OwnedGetObjectFileInfo>,
shared: Option<Arc<GetObjectMetadataCacheEntry>>,
condition_stopped: bool,
}
struct OwnedGetObjectFileInfo {
@@ -940,6 +946,7 @@ impl GetObjectFileInfo {
late_metadata_fanout_disks: None,
}),
shared: None,
condition_stopped: false,
}
}
@@ -957,13 +964,16 @@ impl GetObjectFileInfo {
late_metadata_fanout_disks: Some(late_metadata_fanout_disks),
}),
shared: None,
condition_stopped: false,
}
}
fn shared(entry: Arc<GetObjectMetadataCacheEntry>) -> Self {
let condition_stopped = entry.capability == GetObjectMetadataCacheCapability::ConditionOnly;
Self {
owned: None,
shared: Some(entry),
condition_stopped,
}
}
@@ -1144,6 +1154,248 @@ mod prepared_get_object_metadata_tests {
assert!(take_prepared_get_object_metadata().is_none());
}
struct StopReadForEtag(String);
impl crate::object_api::GetObjectReadCondition for StopReadForEtag {
fn is_terminal(&self, metadata: crate::object_api::GetObjectReadMetadata<'_>) -> bool {
!metadata.delete_marker && metadata.etag == Some(self.0.as_str())
}
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn read_condition_stops_on_quorum_before_slow_disk_and_body_verification() {
use crate::object_api::with_get_object_read_condition;
for size in [4096, 1_300_000] {
let (dirs, set) = make_local_set_disks(4, 2).await;
let bucket = "read-condition-quorum";
let object = object_with_initial_data_shards(bucket, &format!("read-condition-{size}"));
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
set.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let mut put = PutObjReader::from_vec(vec![b'v'; size]);
let info = set.put_object(bucket, &object, &mut put, &opts).await.expect("write object");
for dir in &dirs {
let object_path = dir.path().join(bucket).join(&object);
if size == 4096 {
let path = object_path.join("xl.meta");
let mut bytes = tokio::fs::read(&path).await.expect("read inline metadata");
*bytes.last_mut().expect("inline payload") ^= 1;
tokio::fs::write(path, bytes).await.expect("damage inline data");
} else {
tokio::fs::remove_dir_all(object_path.join(info.data_dir.expect("data directory").to_string()))
.await
.expect("remove shards");
}
}
let slow = bounded_initial_parity_disk_index(bucket, &object);
let barrier = rename_fanout_barrier::arm(&object, slow, rename_fanout_barrier::PHASE_READ_VERSION);
let calls = disk_call_counters::observe(&object);
let read_opts = ObjectOptions {
metadata_cache_safe: true,
..opts
};
let followup_opts = read_opts.clone();
let condition = Arc::new(StopReadForEtag(info.etag.clone().expect("committed ETag")));
let set_for_read = Arc::clone(&set);
let object_for_read = object.clone();
let mut read = tokio::spawn(async move {
with_get_object_read_condition(
condition,
set_for_read.get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &read_opts),
)
.await
});
tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
.await
.expect("slow metadata read is pending");
let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut read)
.await
.expect("condition must not wait for the slow disk or decode damaged data")
.expect("reader task")
.expect("metadata-only result");
assert_eq!(reader.object_info.etag, info.etag);
assert_eq!(
calls.total(disk_call_counters::KIND_INLINE_SHARD_OPEN),
0,
"damaged inline data was never opened"
);
assert_eq!(
calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN),
0,
"missing shards were never opened"
);
assert_eq!(
calls.total(disk_call_counters::KIND_BODY_DECODE),
0,
"the body verifier and decoder were never entered"
);
let mut body = Vec::new();
assert!(
reader.stream.read_to_end(&mut body).await.is_err(),
"terminal metadata cannot be served as a successful body"
);
assert!(body.is_empty());
let first_calls = calls.total(disk_call_counters::KIND_READ_VERSION);
drop(reader);
barrier.release();
let snapshot = set
.get_object_fileinfo_for_get_object_reader(bucket, &object, &followup_opts, true, true)
.await
.expect("fresh metadata after terminal scope");
assert!(!snapshot.condition_stopped);
assert!(
calls.total(disk_call_counters::KIND_READ_VERSION) > first_calls,
"an incomplete terminal snapshot must not satisfy the next metadata read from cache"
);
}
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn read_condition_skips_shard_open_and_decode_with_live_positive_controls() {
for size in [4096, 1_300_000] {
let (_dirs, set) = make_local_set_disks(4, 2).await;
let bucket = "read-condition-body-counters";
let object = object_with_initial_data_shards(bucket, &format!("body-counters-{size}"));
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let payload: Vec<_> = (0..size)
.map(|index| u8::try_from(index % 251).expect("pattern byte"))
.collect();
set.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let mut put = PutObjReader::from_vec(payload.clone());
let info = set.put_object(bucket, &object, &mut put, &opts).await.expect("write object");
let calls = disk_call_counters::observe(&object);
let reader = crate::object_api::with_get_object_read_condition(
Arc::new(StopReadForEtag(info.etag.expect("committed ETag"))),
set.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts),
)
.await
.expect("terminal metadata reader");
drop(reader);
assert!(calls.total(disk_call_counters::KIND_READ_VERSION) > 0, "metadata really was read");
assert_eq!(
calls.total(disk_call_counters::KIND_INLINE_SHARD_OPEN),
0,
"no inline shard source opened"
);
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 0, "no disk shard source opened");
assert_eq!(
calls.total(disk_call_counters::KIND_BODY_DECODE),
0,
"no body verifier or decoder entered"
);
let mut reader = crate::object_api::with_get_object_read_condition(
Arc::new(StopReadForEtag("different-etag".to_owned())),
set.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts),
)
.await
.expect("unmatched condition opens the intact object");
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("positive-control body");
assert_eq!(body, payload);
let source_kind = if size == 4096 {
disk_call_counters::KIND_INLINE_SHARD_OPEN
} else {
disk_call_counters::KIND_DISK_SHARD_OPEN
};
assert!(
calls.total(source_kind) > 0,
"the same source counter observes the positive control: size={size}, inline={}, disk={}, decode={}",
calls.total(disk_call_counters::KIND_INLINE_SHARD_OPEN),
calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN),
calls.total(disk_call_counters::KIND_BODY_DECODE),
);
assert!(
calls.total(disk_call_counters::KIND_BODY_DECODE) > 0,
"the same decode counter observes the positive control"
);
}
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn read_condition_rechecks_selected_metadata_before_body_access() {
struct Changing(std::sync::atomic::AtomicUsize);
impl crate::object_api::GetObjectReadCondition for Changing {
fn is_terminal(&self, _: crate::object_api::GetObjectReadMetadata<'_>) -> bool {
self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed) == 0
}
}
let (_dirs, set) = make_local_set_disks(4, 2).await;
let bucket = "read-condition-consistency";
let object = object_with_initial_data_shards(bucket, "read-condition-consistency");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
set.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let mut put = PutObjReader::from_vec(vec![b'x'; 4096]);
set.put_object(bucket, &object, &mut put, &opts).await.expect("write object");
let result = crate::object_api::with_get_object_read_condition(
Arc::new(Changing(std::sync::atomic::AtomicUsize::new(0))),
set.get_object_reader(bucket, &object, None, HeaderMap::new(), &opts),
)
.await;
assert!(result.is_err(), "an unverified terminal snapshot must never become a 200 reader");
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn read_condition_miss_preserves_inline_early_stop_and_exact_body() {
let (_dirs, set) = make_local_set_disks(4, 2).await;
let bucket = "read-condition-miss";
let object = object_with_initial_data_shards(bucket, "read-condition-miss");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let payload: Vec<u8> = (0..8192)
.map(|index| u8::try_from(index % 251).expect("pattern byte"))
.collect();
set.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let mut put = PutObjReader::from_vec(payload.clone());
set.put_object(bucket, &object, &mut put, &opts)
.await
.expect("write inline object");
let slow = bounded_initial_parity_disk_index(bucket, &object);
let barrier = rename_fanout_barrier::arm(&object, slow, rename_fanout_barrier::PHASE_READ_VERSION);
let set_for_read = Arc::clone(&set);
let object_for_read = object.clone();
let mut read = tokio::spawn(async move {
crate::object_api::with_get_object_read_condition(
Arc::new(StopReadForEtag("other".to_owned())),
set_for_read.get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts),
)
.await
});
tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
.await
.expect("slow metadata read is pending");
let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut read)
.await
.expect("condition miss preserves the normal early stop")
.expect("reader task")
.expect("reader");
let mut bytes = Vec::new();
reader.stream.read_to_end(&mut bytes).await.expect("normal body");
assert_eq!(bytes, payload);
barrier.release();
}
#[test]
fn cache_hit_consumers_release_snapshot_at_legacy_ownership() {
let fi = FileInfo {
@@ -1152,6 +1404,7 @@ mod prepared_get_object_metadata_tests {
};
let cached = Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
capability: GetObjectMetadataCacheCapability::Full,
parts_metadata: vec![fi.clone()],
fi,
online_disks: vec![None],
@@ -4434,6 +4687,13 @@ impl Hash for GetObjectMetadataCacheKey {
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum GetObjectMetadataCacheCapability {
Full,
// Quorum-validated metadata without a complete body or write-target layout.
ConditionOnly,
}
#[derive(Debug)]
struct GetObjectMetadataCacheEntry {
#[allow(dead_code)] // Kept for debugging; moka handles TTL internally
@@ -4442,6 +4702,7 @@ struct GetObjectMetadataCacheEntry {
parts_metadata: Vec<FileInfo>,
online_disks: Vec<Option<DiskStore>>,
read_quorum: usize,
capability: GetObjectMetadataCacheCapability,
}
#[derive(Clone, Debug)]
+59
View File
@@ -2509,6 +2509,22 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
return Err(to_object_err(Error::MethodNotAllowed, vec![bucket, object]));
}
let condition_terminal =
crate::object_api::get_object_read_condition_is_terminal(crate::object_api::GetObjectReadMetadata {
bucket,
object,
etag: object_info.etag.as_deref(),
mod_time: object_info.mod_time,
parts: &object_info.parts,
delete_marker: false,
});
if snapshot.condition_stopped && !condition_terminal {
return Err(Error::other("conditional metadata snapshot no longer satisfies the read condition"));
}
if condition_terminal {
return Ok(GetObjectReader::condition_stopped(object_info));
}
// if object_info.size == 0 {
// let empty_rd: Box<dyn AsyncRead> = Box::new(Bytes::new());
@@ -2608,6 +2624,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
// Decode directly
let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
#[cfg(test)]
crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
if let Some(body) = try_read_inline_data_shards_direct(&mut readers, data_shards, read_length, object_size).await
{
if let Some(decode_stage_start) = decode_stage_start {
@@ -2677,6 +2695,8 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
let decode_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then(Instant::now);
let mut output = Cursor::new(Vec::with_capacity(object_size));
#[cfg(test)]
crate::set_disk::disk_call_counters::record(object, crate::set_disk::disk_call_counters::KIND_BODY_DECODE, 0);
let (written, err) = erasure.decode(&mut output, readers, 0, object_size, object_size).await;
if let Some(e) = err {
return Err(to_object_err(e.into(), vec![bucket, object]));
@@ -15371,6 +15391,7 @@ mod transition_commit_failure_tests {
cache_key.clone(),
Arc::new(GetObjectMetadataCacheEntry {
created_at: Instant::now(),
capability: crate::set_disk::GetObjectMetadataCacheCapability::Full,
fi,
parts_metadata,
online_disks,
@@ -17698,6 +17719,44 @@ mod transition_upload_integrity_tests {
/// Plain objects must keep streaming the remote bytes through untouched:
/// their plan is `Plain`, so the tiered read stays byte-identical.
#[tokio::test]
#[serial_test::serial]
async fn transitioned_conditional_read_skips_remote_get_with_live_positive_control() {
struct MatchingEtag(String);
impl crate::object_api::GetObjectReadCondition for MatchingEtag {
fn is_terminal(&self, metadata: crate::object_api::GetObjectReadMetadata<'_>) -> bool {
!metadata.delete_marker && metadata.etag == Some(self.0.as_str())
}
}
let (_dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "transitioned-condition-body-get";
let object = "object.bin";
let payload = b"conditional remote reads need metadata only".repeat(256);
let original = write_source(&set, &disks, bucket, object, &payload).await;
let tier = format!("CONDITION{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier).await;
set.transition_object(bucket, object, &transition_options(&original, tier))
.await
.expect("transition object");
assert_eq!(backend.get_count().await, 0, "transition setup has not fetched the remote body");
let opts = ObjectOptions {
no_lock: true,
..Default::default()
};
let reader = crate::object_api::with_get_object_read_condition(
Arc::new(MatchingEtag(original.etag.expect("source ETag"))),
set.get_object_reader(bucket, object, None, HeaderMap::new(), &opts),
)
.await
.expect("terminal transitioned metadata");
assert!(reader.object_info.is_remote(), "the condition was evaluated on transitioned metadata");
drop(reader);
assert_eq!(backend.get_count().await, 0, "a matching condition never calls the warm backend GET");
let (body, _) = read_transitioned(&set, bucket, object, None, &opts).await;
assert_eq!(body, payload);
assert_eq!(backend.get_count().await, 1, "the same backend counter observes a normal remote GET");
}
#[tokio::test]
#[serial_test::serial]
async fn transitioned_plain_object_get_is_unchanged() {
File diff suppressed because it is too large Load Diff
+50 -1
View File
@@ -176,7 +176,15 @@ fn every_rule_has_a_positive_sample() {
"heal-no-datadir",
msg("heal: latest metadata for b/o has no data_dir, cannot heal object data"),
),
("heal-all-writes-failed", msg("all drives had write errors, unable to heal b/o")),
(
"heal-all-writes-failed",
Sample {
message: "Set disk object heal writer failures",
level: Some(LogLevel::Warn),
fields: &[("state", "writer_unavailable"), ("result", "all_targets_unavailable")],
..Default::default()
},
),
("heal-rename-failed", msg("heal rename incomplete: 1 of 2 targets committed for b/o")),
(
"heal-xlmeta-regen-failed",
@@ -412,3 +420,44 @@ fn smoke_samples_hit_exact_rule_sets() {
&[],
);
}
#[test]
fn heal_all_writes_failed_matches_current_and_legacy_logs() {
let engine = engine();
let rule_ids = |sample: &Sample| {
engine
.matches(&event(sample))
.into_iter()
.map(|index| engine.rules()[index].id.as_str())
.collect::<Vec<_>>()
};
assert_eq!(
rule_ids(&Sample {
message: "Set disk object heal writer failures",
level: Some(LogLevel::Warn),
fields: &[("state", "writer_unavailable"), ("result", "all_targets_unavailable")],
..Default::default()
}),
vec!["heal-all-writes-failed"]
);
assert_eq!(
rule_ids(&msg("all drives had write errors, unable to heal b/o")),
vec!["heal-all-writes-failed"]
);
for sample in [
Sample {
message: "Set disk object heal writer failures",
fields: &[("state", "writer_unavailable"), ("result", "partial_targets_unavailable")],
..Default::default()
},
msg("Set disk object heal writer failures"),
Sample {
message: "unrelated operation failed",
fields: &[("result", "all_targets_unavailable")],
..Default::default()
},
] {
assert!(rule_ids(&sample).is_empty(), "unexpected rule match: {}", sample.message);
}
}
+1
View File
@@ -18,6 +18,7 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos on storage-sensitive PRs and nightly via `e2e-distributed` |
| Fuzz | `cargo-fuzz` targets over untrusted parsing surfaces; isolated sub-workspace under `fuzz/` | `./scripts/fuzz/run.sh` (see [`fuzz/README.md`](../../fuzz/README.md)) | PR smoke on the paths listed in `.github/workflows/fuzz.yml`, plus nightly corpus |
| Benchmarks | Criterion benches under each crate's `benches/` | `cargo bench -p <crate>` | On demand; never a gate |
| Conditional reads | Paired ordinary/hit/miss GET latency, throughput and read-lock measurements | `scripts/run_conditional_read_abba.py` | Local conditional-read acceptance; [guide and scope](conditional-read-benchmark.md) |
Every script named above is indexed with status and wiring in [`scripts/README.md`](../../scripts/README.md). Fixed GHSA advisories map to named regression tests in [security-regressions.md](security-regressions.md).
@@ -0,0 +1,95 @@
# Conditional object-read benchmark
**Use this when:** validating the performance of metadata-based conditional GET
changes or reproducing a paired application-read regression experiment.
Conditional GET acceptance needs paired performance evidence for ordinary reads,
condition hits, and condition misses. Saving body work on a 304 does not establish
that other reads are unaffected. Keep performance acceptance pending if resource
isolation, baseline stability, or regression budgets fail.
The manually invoked `conditional_read_paired_benchmark` drives the actual
`DefaultObjectUsecase` against four local EC disks. It consumes and verifies the
entire successful body, validates 304 results, and samples existing read-lock
hold histograms. It covers 4 KiB inline and 1.3 MB sharded objects, body cache off
and warmed, and a 20 ms delay on one metadata disk. It excludes network ingress,
authentication, external tiers, cold-fill contention, and distributed capacity.
The test-util-only metadata guard bypasses metadata caching in the dedicated
`conditional-bench` bucket so cache hits cannot hide the slow-disk scenario.
Both ordinary GET and prepared GET metadata reads already use `read_data=true`
and therefore exercise the existing delay injector without an additional override.
The guard has no effect in ordinary builds or outside its bucket, and ownership
drop restores the default policy. Both compared revisions must include this
identical benchmark control.
## Build paired artifacts
Use a clean detached worktree at the full baseline commit, a separate candidate
worktree, and matching stable toolchain, features, and build profile. The adapter
resolves every source anchor before writing. It installs only the test harness,
test-util controls, and their facade exports; it does not install the conditional
read implementation on the baseline. Review its diff before building.
```bash
python3 scripts/prepare_conditional_read_baseline.py \
--baseline-worktree "$BASELINE_WORKTREE" --revision "$BASELINE_REVISION"
```
In each worktree, format the adapter and build the same artifact:
```bash
RUSTUP_TOOLCHAIN=stable cargo fmt --all
RUSTUP_TOOLCHAIN=stable cargo test --locked -p rustfs --lib \
--no-default-features --features rio-v2 \
conditional_read_paired_benchmark --no-run
```
Save each executable printed by Cargo before another build overwrites it. For
deployment acceptance, use matching optimized profiles and an isolated target
environment. A test-profile run is a local application regression experiment,
not a production throughput estimate.
## Run and interpret
Stop concurrent builds and other workloads on the measurement resources. The
runner refuses concurrent Cargo/rustc processes or high host load. It executes
fresh processes serially in A1–B1–B2–A2 order, removes ambient RustFS experimental
settings and proxies, uses independent temporary disks, and records binary,
revision, harness, and profile provenance beside raw logs.
Executable snapshots are read-only and their hashes are rechecked before and
after every case. Cargo/rustc activity is sampled during each process; host load
and isolation are checked again on exit, including the last A2 case. These checks
do not replace exclusive resource ownership for formal deployment acceptance.
```bash
python3 scripts/run_conditional_read_abba.py \
--baseline "$BASELINE_BINARY" --candidate "$CANDIDATE_BINARY" \
--baseline-revision "$BASELINE_REVISION" \
--candidate-revision "$CANDIDATE_REVISION" \
--harness rustfs/src/app/object/conditional_read_bench.rs \
--profile test --out-dir "$RESULT_DIR"
```
The default is 1000 requests per case at concurrency 8. Each case records
throughput, p95/p99 latency, read-lock p95/p99, and a zero metadata-cache-hit
assertion. Body-cache-on cases are warmed; they do not measure cold-fill misses.
Inline shard bytes can be carried by an `xl.meta` read even when no separate
body reader, bitrot verification, or decode is needed.
Budgets are fixed before the run:
- Baseline drift: throughput within 10%; latency and lock percentiles within
15%, with a 0.25 ms absolute noise floor.
- Candidate regression: throughput loss at most 5%; latency and lock percentile
increases at most 10%, with a 0.25 ms absolute noise floor.
A drift failure produces no candidate attribution. After a stable baseline, the
runner compares geometric means of the paired legs. Any workload over budget
keeps acceptance pending; improving 304 cannot compensate for an ordinary-read
or condition-miss regression. Preserve raw logs and investigate before rerunning;
do not relax budgets after seeing the results.
Run `python3 scripts/test_conditional_read_abba.py` to verify the acceptance
parser and drift/regression gates.
+46 -1
View File
@@ -271,7 +271,7 @@ async fn call_website(
website_error_with_document(&req, &server_ctx, &target.bucket, config.error_document.as_ref(), StatusCode::NOT_FOUND)
.await
}
Err(err) if err.code() == &S3ErrorCode::NotModified => Ok(website_error(StatusCode::NOT_MODIFIED, true)),
Err(err) if err.code() == &S3ErrorCode::NotModified => Ok(website_not_modified(&err)),
Err(_) if status.is_client_error() => {
website_error_with_document(&req, &server_ctx, &target.bucket, config.error_document.as_ref(), status).await
}
@@ -520,6 +520,18 @@ fn website_error(status: StatusCode, head: bool) -> S3Response<Body> {
}
}
fn website_not_modified(error: &S3Error) -> S3Response<Body> {
let mut response = S3Response::new(Body::empty());
response.status = Some(StatusCode::NOT_MODIFIED);
if let Some(headers) = error.headers() {
response.headers = headers.clone();
}
response.headers.remove(http::header::CONTENT_LENGTH);
response.headers.remove(http::header::CONTENT_TYPE);
response.headers.remove(http::header::TRANSFER_ENCODING);
response
}
fn website_location(location: String, status: StatusCode) -> S3Result<S3Response<Body>> {
let mut response = S3Response::new(Body::empty());
response.status = Some(status);
@@ -826,6 +838,39 @@ fn header_value(headers: &HeaderMap, name: &str) -> S3Result<Option<String>> {
#[cfg(test)]
mod tests {
#[tokio::test]
async fn website_304_preserves_cache_headers_without_content() {
use http_body_util::BodyExt;
let mut error = s3s::S3Error::new(s3s::S3ErrorCode::NotModified);
let mut headers = http::HeaderMap::new();
for (name, value) in [
("etag", "\"abc\""),
("cache-control", "max-age=60"),
("expires", "Tue, 14 Nov 2023 22:13:20 GMT"),
("content-length", "123"),
("content-type", "application/xml"),
] {
headers.insert(name, http::HeaderValue::from_static(value));
}
error.set_headers(headers);
let response = super::website_not_modified(&error);
assert_eq!(response.status, Some(http::StatusCode::NOT_MODIFIED));
assert_eq!(response.headers["etag"], "\"abc\"");
assert_eq!(response.headers["cache-control"], "max-age=60");
assert!(response.headers.contains_key("expires"));
assert!(!response.headers.contains_key("content-length"));
assert!(!response.headers.contains_key("content-type"));
assert!(
response
.output
.collect()
.await
.expect("empty website body")
.to_bytes()
.is_empty()
);
}
use super::{
MetadataOperation, list_object_versions_input, list_objects_v2_input, metadata_operation, website_redirect,
website_relative_redirect, website_rule_matches, website_target,
@@ -0,0 +1,148 @@
// 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.
//! Manually invoked, paired application-read benchmark on real local EC disks.
//! Run the identical harness on both revisions in separate processes. Environment
//! configuration must be set before either process initializes the store.
use super::*;
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
use futures::stream;
use http::{HeaderValue, Method};
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
use std::time::Instant;
fn percentile(samples: &mut [f64], percent: usize) -> f64 {
assert!(!samples.is_empty(), "benchmark must collect samples");
samples.sort_by(f64::total_cmp);
samples[(samples.len() * percent).div_ceil(100).saturating_sub(1)]
}
async fn read_once(usecase: &DefaultObjectUsecase, object: &str, etag: &str, kind: &str, expected: &[u8]) -> f64 {
let start = Instant::now();
let mut req = build_request(
GetObjectInput::builder()
.bucket("conditional-bench".to_owned())
.key(object.to_owned())
.build()
.expect("benchmark input"),
Method::GET,
);
if kind != "unconditional" {
let tag = if kind == "hit" { etag } else { "unmatched-benchmark-etag" };
req.headers.insert(
http::header::IF_NONE_MATCH,
HeaderValue::from_str(&format!("\"{tag}\"")).expect("benchmark condition"),
);
}
match usecase.execute_get_object(req).await {
Err(error) => {
assert_eq!(kind, "hit");
assert_eq!(error.code(), &S3ErrorCode::NotModified);
}
Ok(mut response) => {
assert_ne!(kind, "hit");
let mut body = response.output.body.take().expect("benchmark body");
let mut actual = Vec::with_capacity(expected.len());
while let Some(chunk) = body.next().await {
actual.extend_from_slice(&chunk.expect("benchmark body chunk"));
}
assert_eq!(actual, expected, "benchmark must consume and validate the whole body");
}
}
start.elapsed().as_secs_f64() * 1000.0
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "manual benchmark: use the paired runner with --ignored --nocapture"]
async fn conditional_read_paired_benchmark() {
let iterations = std::env::var("RUSTFS_CONDITIONAL_BENCH_ITERATIONS")
.unwrap_or_else(|_| "1000".to_owned())
.parse::<usize>()
.expect("benchmark iteration count");
let concurrency = std::env::var("RUSTFS_CONDITIONAL_BENCH_CONCURRENCY")
.unwrap_or_else(|_| "8".to_owned())
.parse::<usize>()
.expect("benchmark concurrency");
assert!(iterations >= 100 && concurrency > 0);
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::set_global_recorder(recorder).expect("run the benchmark alone in a fresh process");
let _metadata_profile = crate::app::storage_api::test::set_disk::ConditionalReadBenchmarkMetadataGuard::acquire();
let (store, context) = real_store_test_context().await;
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
let cache_enabled = !context.object_data_cache().is_disabled();
let usecase = DefaultObjectUsecase::with_context(Some(context));
store
.make_bucket("conditional-bench", &MakeBucketOptions::default())
.await
.expect("benchmark bucket");
for size in [4096, 1_300_000] {
let object = format!("bench-{size}");
let expected = (0..size).map(|index| (index % 251) as u8).collect::<Vec<_>>();
let info = put_real_cold_fill_object(&store, "conditional-bench", &object, &expected).await;
let etag = info.etag.as_deref().expect("benchmark ETag");
for kind in ["unconditional", "hit", "miss"] {
// Prime the body cache and filesystem for every case, including 304.
for _ in 0..20 {
read_once(&usecase, &object, etag, "unconditional", &expected).await;
}
snapshotter.snapshot();
let start = Instant::now();
let mut latencies = stream::iter(0..iterations)
.map(|_| read_once(&usecase, &object, etag, kind, &expected))
.buffer_unordered(concurrency)
.collect::<Vec<_>>()
.await;
let elapsed = start.elapsed().as_secs_f64();
let mut lock_ms = Vec::new();
let mut metadata_cache_hits = 0;
for (key, _, _, value) in snapshotter.snapshot().into_vec() {
if key.key().name() == "rustfs_io_get_object_metadata_cache_total"
&& key
.key()
.labels()
.any(|label| label.key() == "decision" && label.value() == "hit")
&& let DebugValue::Counter(count) = &value
{
metadata_cache_hits += count;
}
if key.key().name() == "rustfs_object_lock_diag_hold_duration_seconds"
&& key
.key()
.labels()
.any(|label| label.key() == "mode" && label.value() == "read")
&& let DebugValue::Histogram(values) = value
{
lock_ms.extend(values.iter().map(|sample| sample.0 * 1000.0));
}
}
assert!(!lock_ms.is_empty(), "enable RUSTFS_OBJECT_LOCK_DIAG_ENABLE before startup");
assert_eq!(metadata_cache_hits, 0, "slow-disk measurements require a metadata-cache bypass");
println!(
"CONDITIONAL_BENCH {}",
serde_json::json!({
"size": size, "kind": kind, "cache": cache_enabled,
"iterations": iterations, "concurrency": concurrency,
"metadata_cache_hits": metadata_cache_hits, "metadata_cache_bypassed": true,
"slowtail_ms": std::env::var("RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS").unwrap_or_default(),
"ops_per_sec": iterations as f64 / elapsed,
"p95_ms": percentile(&mut latencies, 95), "p99_ms": percentile(&mut latencies, 99),
"lock_samples": lock_ms.len(), "lock_p95_ms": percentile(&mut lock_ms, 95),
"lock_p99_ms": percentile(&mut lock_ms, 99),
})
);
}
}
}
+523 -77
View File
@@ -15,7 +15,30 @@
//! GetObject / GetObjectAttributes read path: cold fill, resume, stream tuning.
use super::*;
use crate::app::storage_api::object_usecase::object_cache::{
GetObjectReadCondition, GetObjectReadMetadata, evaluate_read_preconditions, has_read_preconditions,
with_get_object_read_condition,
};
use crate::on_demand_migration::WriteBackBody;
struct ConditionalRead {
bucket: String,
object: String,
headers: HeaderMap,
part_number: Option<usize>,
}
impl GetObjectReadCondition for ConditionalRead {
fn is_terminal(&self, metadata: GetObjectReadMetadata<'_>) -> bool {
if metadata.bucket != self.bucket || metadata.object != self.object || metadata.delete_marker {
return false;
}
evaluate_read_preconditions(&self.headers, metadata.etag, metadata.mod_time).is_err()
|| self
.part_number
.is_some_and(|number| number > 1 && !metadata.parts.iter().any(|part| part.number == number))
}
}
use crate::on_demand_migration::{
BucketOdmState, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, PullError, PullLeader, PullOutcome, PullReason, PullSlot,
RangeGetPolicy, SourceBody, SourceClient, SourceError, SourceGet, SourceHead, commit_inline, idle_guarded_body,
@@ -49,6 +72,39 @@ type ColdFillReaderOpenProbeState = Option<(rustfs_object_data_cache::ObjectData
#[cfg(test)]
static COLD_FILL_READER_OPEN_PROBE: OnceLock<Mutex<ColdFillReaderOpenProbeState>> = OnceLock::new();
#[cfg(test)]
#[derive(Default)]
struct BodyCacheFillCalls {
reservations: AtomicU64,
fills: AtomicU64,
}
#[cfg(test)]
type BodyCacheFillProbeState = Option<(rustfs_object_data_cache::ObjectDataCacheKey, Arc<BodyCacheFillCalls>)>;
#[cfg(test)]
static BODY_CACHE_FILL_PROBE: OnceLock<Mutex<BodyCacheFillProbeState>> = OnceLock::new();
#[cfg(test)]
fn record_body_cache_fill_call_for_test(plan: &rustfs_object_data_cache::ObjectDataCacheGetPlan, reservation: bool) {
let Some(key) = plan.key() else { return };
let counters = BODY_CACHE_FILL_PROBE
.get_or_init(|| Mutex::new(None))
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.as_ref()
.filter(|(probe_key, _)| probe_key == key)
.map(|(_, counts)| Arc::clone(counts));
if let Some(counters) = counters {
let counter = if reservation {
&counters.reservations
} else {
&counters.fills
};
counter.fetch_add(1, Ordering::Relaxed);
}
}
fn adjust_cold_fill_disk_permit_metric(owner: ColdFillDiskPermitOwner, acquired: bool) {
macro_rules! adjust_gauge {
($name:literal) => {{
@@ -1075,6 +1131,8 @@ async fn start_cold_fill_producer<AcquireIo, AcquireIoFuture, OpenReader, OpenRe
let publish = async {
#[cfg(test)]
wait_cold_fill_publication_barrier(&engine_plan).await;
#[cfg(test)]
record_body_cache_fill_call_for_test(&engine_plan, false);
adapter.fill_reserved_body(&engine_plan, reserved).await
};
tokio::pin!(publish);
@@ -2106,6 +2164,42 @@ fn should_buffer_get_object_in_memory_with_threshold(
}
impl DefaultObjectUsecase {
fn with_get_object_conditions<F: std::future::Future>(
req: &S3Request<GetObjectInput>,
part_number: Option<usize>,
read: F,
) -> impl std::future::Future<Output = F::Output> {
// The concrete reader future is large; keep it out of the request's stack frame.
let read = Box::pin(read);
let condition = if has_read_preconditions(&req.headers) {
let mut headers = HeaderMap::with_capacity(4);
for name in [
http::header::IF_MATCH,
http::header::IF_UNMODIFIED_SINCE,
http::header::IF_NONE_MATCH,
http::header::IF_MODIFIED_SINCE,
] {
if let Some(value) = req.headers.get(&name) {
headers.insert(name, value.clone());
}
}
Some(Arc::new(ConditionalRead {
bucket: req.input.bucket.clone(),
object: rustfs_utils::path::encode_dir_object_ref(&req.input.key).into_owned(),
headers,
part_number,
}))
} else {
None
};
async move {
if let Some(condition) = condition {
with_get_object_read_condition(condition, read).await
} else {
read.await
}
}
}
fn build_memory_bytes_blob(
bytes: Bytes,
response_content_length: i64,
@@ -2279,9 +2373,29 @@ impl DefaultObjectUsecase {
Ok(())
}
fn validate_get_object_before_cold_fill(headers: &HeaderMap, part_number: Option<usize>, info: &ObjectInfo) -> S3Result<()> {
check_preconditions(headers, info)?;
Self::validate_get_object_part_number(part_number, info)
async fn validate_get_object_before_body(
req: &S3Request<GetObjectInput>,
bucket: &str,
key: &str,
part_number: Option<usize>,
info: &ObjectInfo,
) -> S3Result<()> {
if info.delete_marker {
let code = if req.input.version_id.is_some() {
S3ErrorCode::MethodNotAllowed
} else {
S3ErrorCode::NoSuchKey
};
return Err(with_delete_marker_read_headers(S3Error::new(code), info, req.input.version_id.as_deref()));
}
let result =
check_preconditions(&req.headers, info).and_then(|()| Self::validate_get_object_part_number(part_number, info));
if result.is_err() {
let principal = SseKmsPrincipal::from_request(req);
crate::storage::sse::validate_conditional_read(bucket, key, &info.user_defined, &req.headers, principal.as_ref())
.await?;
}
result
}
/// The `x-amz-mp-parts-count` value for a response, if one is owed.
@@ -2524,6 +2638,7 @@ impl DefaultObjectUsecase {
let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start);
let store_headers = project_ssec_transport_headers(&req.headers);
let cache_adapter = self.object_data_cache();
let conditional = has_read_preconditions(&req.headers);
if cache_adapter.is_disabled() || !cache_adapter.materialize_fill_enabled() {
let io_planning = Self::acquire_get_object_io_planning(
manager,
@@ -2537,7 +2652,11 @@ impl DefaultObjectUsecase {
.await?;
let reader = track_object_read_setup(
object_traffic_health.as_deref(),
store.get_object_reader(bucket, key, rs.clone(), store_headers, opts),
Self::with_get_object_conditions(
req,
part_number,
store.get_object_reader(bucket, key, rs.clone(), store_headers, opts),
),
)
.await
.map_err(map_get_object_reader_error)?;
@@ -2560,16 +2679,18 @@ impl DefaultObjectUsecase {
)
.await?,
);
let mut prepared = Some(
track_object_read_setup(
object_traffic_health.as_deref(),
store.prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts),
)
.await
.map_err(map_get_object_reader_error)?,
);
let initial_metadata = track_object_read_setup(
object_traffic_health.as_deref(),
store.prepare_get_object_reader(bucket, key, rs.clone(), HeaderMap::new(), opts),
)
.await
.map_err(map_get_object_reader_error)?;
let mut cache_fill_allowed = true;
let mut legacy_hook_missed = false;
if conditional {
Self::validate_get_object_before_body(req, bucket, key, part_number, initial_metadata.object_info()).await?;
}
let mut prepared = Some(initial_metadata);
'snapshot: {
let info = prepared
.as_ref()
@@ -2609,7 +2730,6 @@ impl DefaultObjectUsecase {
if matches!(legacy_probe, GetObjectBodyCacheHookLookup::Ineligible) {
break 'snapshot;
}
Self::validate_get_object_before_cold_fill(&req.headers, part_number, info)?;
if let GetObjectBodyCacheHookLookup::Hit(body) = legacy_probe {
drop(metadata_admission.take());
let info = prepared
@@ -2770,6 +2890,8 @@ impl DefaultObjectUsecase {
return;
};
#[cfg(test)]
record_body_cache_fill_call_for_test(&engine_plan, true);
let reservation = adapter.reserve_body(&engine_plan);
#[cfg(test)]
let reader_open_plan = engine_plan.clone();
@@ -2826,10 +2948,12 @@ impl DefaultObjectUsecase {
let io_planning = metadata_admission
.take()
.ok_or_else(|| s3_error!(InternalError, "prepared metadata admission is unavailable"))?;
let reader =
track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader())
.await
.map_err(map_get_object_reader_error)?;
let reader = track_object_read_setup(
object_traffic_health.as_deref(),
Self::with_get_object_conditions(req, part_number, prepared.with_headers(store_headers).into_reader()),
)
.await
.map_err(map_get_object_reader_error)?;
(io_planning, reader)
} else {
let io_planning = Self::acquire_get_object_io_planning(
@@ -2849,13 +2973,20 @@ impl DefaultObjectUsecase {
)
.await
.map_err(map_get_object_reader_error)?;
track_object_read_setup(object_traffic_health.as_deref(), prepared.with_headers(store_headers).into_reader())
.await
.map_err(map_get_object_reader_error)?
track_object_read_setup(
object_traffic_health.as_deref(),
Self::with_get_object_conditions(req, part_number, prepared.with_headers(store_headers).into_reader()),
)
.await
.map_err(map_get_object_reader_error)?
} else {
track_object_read_setup(
object_traffic_health.as_deref(),
store.get_object_reader(bucket, key, rs.clone(), store_headers, opts),
Self::with_get_object_conditions(
req,
part_number,
store.get_object_reader(bucket, key, rs.clone(), store_headers, opts),
),
)
.await
.map_err(map_get_object_reader_error)?
@@ -2900,6 +3031,7 @@ impl DefaultObjectUsecase {
let stream = reader.stream;
let buffered_body = reader.buffered_body;
Self::validate_get_object_before_body(req, bucket, key, part_number, &info).await?;
let read_duration = read_start.elapsed();
// Conditional metrics recording to reduce overhead
@@ -2909,9 +3041,6 @@ impl DefaultObjectUsecase {
manager.record_disk_operation(info.size as u64, read_duration, true).await;
}
check_preconditions(&req.headers, &info)?;
Self::validate_get_object_part_number(part_number, &info)?;
debug!(object_size = info.size, part_count = info.parts.len(), "GET object metadata snapshot");
for part in info.parts.iter() {
debug!(
@@ -3411,9 +3540,17 @@ impl DefaultObjectUsecase {
let cache_plan = cache_plan.clone();
let fill_bytes = buffered_body.clone();
tokio::spawn(async move {
#[cfg(test)]
if let GetObjectBodyCachePlan::Cacheable(plan) = &cache_plan {
record_body_cache_fill_call_for_test(plan, false);
}
let _ = fill_get_object_body_cache_from_buffered_body(&cache_adapter, &cache_plan, &fill_bytes).await;
});
} else if cache_fill_allowed {
#[cfg(test)]
if let GetObjectBodyCachePlan::Cacheable(plan) = &cache_plan {
record_body_cache_fill_call_for_test(plan, false);
}
let _ = fill_get_object_body_cache_from_buffered_body(cache_adapter, &cache_plan, &buffered_body).await;
}
@@ -3482,6 +3619,10 @@ impl DefaultObjectUsecase {
let cache_plan = cache_plan.clone();
let fill_bytes = bytes.clone();
tokio::spawn(async move {
#[cfg(test)]
if let GetObjectBodyCachePlan::Cacheable(plan) = &cache_plan {
record_body_cache_fill_call_for_test(plan, false);
}
let _ = fill_get_object_body_cache_from_materialized_body(&cache_adapter, &cache_plan, &fill_bytes).await;
});
@@ -5365,7 +5506,11 @@ mod on_demand_migration_tests {
let data = payload(512);
let etag = source_head(&data).etag.expect("etag");
let source = ScriptedSource::for_object(&data);
let head = source_head(&data);
let source = ScriptedSource::new(
vec![Ok(head.clone()), Ok(head.clone()), Ok(head.clone())],
vec![Ok((head, data.clone(), None))],
);
let mut headers = HeaderMap::new();
headers.insert(http::header::IF_NONE_MATCH, HeaderValue::from_str(&format!("\"{etag}\"")).unwrap());
let err = failed(odm_get_from_source(&state, &source, &headers, KEY, None, None).await);
@@ -5373,7 +5518,6 @@ mod on_demand_migration_tests {
assert_eq!(source.head_calls(), 1);
assert_eq!(source.get_calls(), 0, "a 304 never pulls");
let source = ScriptedSource::for_object(&data);
let mut headers = HeaderMap::new();
headers.insert(http::header::IF_MATCH, HeaderValue::from_static("\"another-etag\""));
let err = failed(odm_get_from_source(&state, &source, &headers, KEY, None, None).await);
@@ -5382,6 +5526,15 @@ mod on_demand_migration_tests {
assert_eq!(get_count(&state, OdmOutcome::SourceHit), 2);
assert_eq!(state.inflight_keys(), 0);
assert!(rt.write_back.puts().is_empty());
let (mut output, _) = served(odm_get_from_source(&state, &source, &HeaderMap::new(), KEY, None, None).await);
assert_eq!(
collect_body(&mut output).await,
data,
"positive control still fetches the source representation"
);
assert_eq!(source.get_calls(), 1, "the same SourceClient counter observes the positive-control GET");
assert_eq!(source.head_calls(), 3, "each conditional and ordinary request reads source metadata");
}
#[tokio::test]
@@ -5671,6 +5824,28 @@ mod tests {
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
#[tokio::test]
async fn conditional_reader_scope_keeps_large_futures_off_the_request_stack() {
let mut req = build_request(GetObjectInput::default(), Method::GET);
for conditional in [false, true] {
if conditional {
req.headers
.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("\"etag\""));
}
let payload = [0u8; 64 * 1024];
let read = async move {
std::hint::black_box(payload);
std::future::ready(()).await;
};
let read = DefaultObjectUsecase::with_get_object_conditions(&req, None, read);
assert!(
std::mem::size_of_val(&read) < 2048,
"reader scope must not embed the concrete reader future"
);
read.await;
}
}
#[tokio::test(start_paused = true)]
async fn cold_fill_disk_admission_preserves_slow_down() {
let manager = Box::leak(Box::new(ConcurrencyManager::with_disk_read_caps_for_test(1, 1)));
@@ -5690,6 +5865,184 @@ mod tests {
drop(primary);
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn conditional_get_304_uses_metadata_without_decoding_damaged_data() {
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
let (store, ambient) = real_cold_fill_test_context().await;
let context = temp_env::with_vars([(rustfs_config::ENV_OBJECT_DATA_CACHE_ENABLE, Some("false"))], || {
Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()))
});
assert!(context.object_data_cache().is_disabled());
let usecase = DefaultObjectUsecase::with_context(Some(context));
let (disk_paths, _) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await;
let bucket = format!("conditional-metadata-{}", Uuid::new_v4());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create test bucket");
for (object, length) in [("inline.bin", 4096), ("shards.bin", 1_300_000)] {
let payload = vec![b'c'; length];
let info = put_real_cold_fill_object(&store, &bucket, object, &payload).await;
let etag = info.etag.as_deref().expect("committed ETag");
for disk in &disk_paths {
let object_path = disk.join(&bucket).join(object);
if length == 4096 {
let meta_path = object_path.join("xl.meta");
let mut metadata = tokio::fs::read(&meta_path).await.expect("read inline metadata");
*metadata.last_mut().expect("inline data bytes") ^= 1;
tokio::fs::write(&meta_path, metadata).await.expect("damage inline payload");
} else {
let data_path = object_path.join(info.data_dir.expect("sharded object data directory").to_string());
tokio::fs::remove_dir_all(data_path).await.expect("remove object data shards");
}
}
let resolved = store
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.expect("metadata remains readable");
assert_eq!(resolved.etag.as_deref(), Some(etag));
let mut req = build_request(
GetObjectInput::builder()
.bucket(bucket.clone())
.key(object.to_owned())
.build()
.expect("GET input"),
Method::GET,
);
req.headers.insert(
http::header::IF_NONE_MATCH,
HeaderValue::from_str(&format!("W/\"{etag}\"")).expect("ETag header"),
);
let error = usecase
.execute_get_object(req)
.await
.expect_err("matching condition returns 304");
assert_eq!(error.code(), &S3ErrorCode::NotModified);
assert_eq!(error.headers().expect("304 validators")["etag"], format!("\"{etag}\""));
let mut req = build_request(
HeadObjectInput::builder()
.bucket(bucket.clone())
.key(object.to_owned())
.build()
.expect("HEAD input"),
Method::HEAD,
);
req.headers.insert(
http::header::IF_NONE_MATCH,
HeaderValue::from_str(&format!("\"{etag}\"")).expect("ETag header"),
);
let error = usecase
.execute_head_object(req)
.await
.expect_err("conditional HEAD returns 304");
assert_eq!(error.code(), &S3ErrorCode::NotModified);
assert_eq!(error.headers().expect("HEAD validators")["etag"], format!("\"{etag}\""));
let req = build_request(
GetObjectInput::builder()
.bucket(bucket.clone())
.key(object.to_owned())
.build()
.expect("GET input"),
Method::GET,
);
if let Ok(mut response) = usecase.execute_get_object(req).await {
let mut body = response.output.body.take().expect("body for damaged GET");
let mut failed = false;
while let Some(chunk) = body.next().await {
if chunk.is_err() {
failed = true;
break;
}
}
assert!(failed, "ordinary GET must still reject damaged data");
}
}
}
#[tokio::test]
async fn conditional_read_checks_encryption_before_not_modified() {
let mut info = ObjectInfo {
etag: Some("abc".to_owned()),
..Default::default()
};
info.user_defined = std::collections::HashMap::from([(
"x-amz-server-side-encryption-customer-algorithm".to_owned(),
"AES256".to_owned(),
)])
.into();
let mut req = build_request(
GetObjectInput::builder()
.bucket("bucket".to_owned())
.key("key".to_owned())
.build()
.expect("GET input"),
Method::GET,
);
req.headers
.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("\"abc\""));
let error = DefaultObjectUsecase::validate_get_object_before_body(&req, "bucket", "key", None, &info)
.await
.expect_err("SSE-C key is mandatory before 304");
assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn conditional_get_miss_preserves_full_and_range_bodies() {
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
let (store, ambient) = real_cold_fill_test_context().await;
let context = temp_env::with_vars([(rustfs_config::ENV_OBJECT_DATA_CACHE_ENABLE, Some("false"))], || {
Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()))
});
let usecase = DefaultObjectUsecase::with_context(Some(context));
let bucket = format!("conditional-miss-{}", Uuid::new_v4());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let payload: Vec<u8> = (0..1_300_000)
.map(|index| u8::try_from(index % 251).expect("pattern byte"))
.collect();
put_real_cold_fill_object(&store, &bucket, "object", &payload).await;
for ranged in [false, true] {
let input = GetObjectInput::builder()
.bucket(bucket.clone())
.key("object".to_owned())
.range(ranged.then_some(Range::Int {
first: 10,
last: Some(29),
}))
.build()
.expect("GET input");
let mut req = build_request(input, Method::GET);
req.headers
.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("\"other\""));
if ranged {
req.headers
.insert(http::header::RANGE, HeaderValue::from_static("bytes=10-29"));
}
let mut response = usecase
.execute_get_object(req)
.await
.expect("unmatched condition serves body");
let expected = if ranged { &payload[10..30] } else { payload.as_slice() };
assert_eq!(
response.output.content_length,
Some(i64::try_from(expected.len()).expect("test body length"))
);
assert_eq!(response.output.content_range.as_deref(), ranged.then_some("bytes 10-29/1300000"));
let mut body = response.output.body.take().expect("GET body");
let mut bytes = Vec::new();
while let Some(chunk) = body.next().await {
bytes.extend_from_slice(&chunk.expect("body chunk"));
}
assert_eq!(bytes, expected);
}
}
#[tokio::test]
async fn cold_fill_closed_disk_admission_is_not_slow_down() {
let manager = Box::leak(Box::new(ConcurrencyManager::with_disk_read_caps_for_test(1, 1)));
@@ -6665,6 +7018,100 @@ mod tests {
assert_eq!(coordinator.active_session_count_for_test(), 0);
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn conditional_get_terminal_response_skips_cache_fill_calls_with_live_positive_control() {
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
struct ResetProbes;
impl Drop for ResetProbes {
fn drop(&mut self) {
*BODY_CACHE_FILL_PROBE
.get_or_init(|| Mutex::new(None))
.lock()
.expect("cache-fill probe") = None;
*COLD_FILL_READER_OPEN_PROBE
.get_or_init(|| Mutex::new(None))
.lock()
.expect("reader-open probe") = None;
}
}
let (store, context) = real_cold_fill_test_context().await;
let bucket = format!("conditional-fill-calls-{}", Uuid::new_v4());
let object = "object.bin";
let payload: Vec<_> = (0..1_300_000)
.map(|index| u8::try_from(index % 251).expect("pattern byte"))
.collect();
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let info = put_real_cold_fill_object(&store, &bucket, object, &payload).await;
let adapter = context.object_data_cache();
let plan = real_cold_fill_plan(&adapter, &bucket, object, &info);
let key = plan.key().expect("cacheable key").clone();
let fills = Arc::new(BodyCacheFillCalls::default());
let reader_opens = Arc::new(AtomicU64::new(0));
let _reset = ResetProbes;
*BODY_CACHE_FILL_PROBE
.get_or_init(|| Mutex::new(None))
.lock()
.expect("cache-fill probe") = Some((key.clone(), Arc::clone(&fills)));
*COLD_FILL_READER_OPEN_PROBE
.get_or_init(|| Mutex::new(None))
.lock()
.expect("reader-open probe") = Some((key, Arc::clone(&reader_opens)));
let usecase = DefaultObjectUsecase::with_context(Some(context));
let input = GetObjectInput::builder()
.bucket(bucket)
.key(object.to_owned())
.build()
.expect("GET input");
for (header, value, expected) in [
(
http::header::IF_NONE_MATCH,
format!("\"{}\"", info.etag.as_deref().expect("ETag")),
S3ErrorCode::NotModified,
),
(http::header::IF_MATCH, "\"other-etag\"".to_owned(), S3ErrorCode::PreconditionFailed),
] {
let mut req = build_request(input.clone(), Method::GET);
req.headers.insert(header, HeaderValue::from_str(&value).expect("condition"));
let error = usecase.execute_get_object(req).await.expect_err("terminal condition");
assert_eq!(error.code(), &expected);
assert_eq!(reader_opens.load(Ordering::Relaxed), 0, "no cold-fill body factory call");
assert_eq!(fills.reservations.load(Ordering::Relaxed), 0, "no cache-fill reservation call");
assert_eq!(fills.fills.load(Ordering::Relaxed), 0, "no cache-fill publish call");
assert_eq!(adapter.cold_fill_coordinator().active_session_count_for_test(), 0);
}
let mut response = usecase
.execute_get_object(build_request(input, Method::GET))
.await
.expect("positive-control GET");
let mut body = response.output.body.take().expect("body");
let mut actual = Vec::new();
while let Some(chunk) = body.next().await {
actual.extend_from_slice(&chunk.expect("positive-control chunk"));
}
assert_eq!(actual, payload);
assert!(
reader_opens.load(Ordering::Relaxed) > 0,
"the same factory probe observes the positive control"
);
assert!(
fills.reservations.load(Ordering::Relaxed) > 0,
"the same reservation probe observes the positive control"
);
assert!(
fills.fills.load(Ordering::Relaxed) > 0,
"the same fill probe observes the positive control"
);
assert!(
matches!(adapter.lookup_body(&plan).await, rustfs_object_data_cache::ObjectDataCacheLookup::Hit(_)),
"the positive control really populated the cache"
);
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn execute_get_object_maps_cold_fill_session_rejection_to_slow_down_without_opening_reader() {
@@ -6732,6 +7179,16 @@ mod tests {
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn execute_get_object_generation_change_bypasses_old_cold_fill_plan() {
run_conditional_generation_change(false).await;
}
#[tokio::test]
#[serial_test::serial(body_cache_hook)]
async fn conditional_get_rechecks_generation_before_fallback_body() {
run_conditional_generation_change(true).await;
}
async fn run_conditional_generation_change(conditional: bool) {
use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
let (store, context) = real_cold_fill_test_context().await;
@@ -6743,6 +7200,16 @@ mod tests {
.expect("real cold-fill generation bucket must be created");
let initial_body = vec![b'a'; 1_300_000];
let changed_body = vec![b'b'; initial_body.len()];
let condition_etag = if conditional {
Some(
put_real_cold_fill_object(&store, &bucket, object, &changed_body)
.await
.etag
.expect("replacement ETag"),
)
} else {
None
};
let initial_info = put_real_cold_fill_object(&store, &bucket, object, &initial_body).await;
let adapter = context.object_data_cache();
let initial_plan = real_cold_fill_plan(&adapter, &bucket, object, &initial_info);
@@ -6769,7 +7236,14 @@ mod tests {
..GetObjectTimeoutPolicy::default()
},
);
let request = tokio::spawn(async move { usecase.execute_get_object(build_request(input, Method::GET)).await });
let mut req = build_request(input, Method::GET);
if let Some(etag) = &condition_etag {
req.headers.insert(
http::header::IF_NONE_MATCH,
HeaderValue::from_str(&format!("\"{etag}\"")).expect("replacement condition"),
);
}
let request = tokio::spawn(async move { usecase.execute_get_object(req).await });
tokio::time::timeout(Duration::from_secs(2), async {
while coordinator.global_waiter_count_for_test() != 1 {
tokio::task::yield_now().await;
@@ -6780,13 +7254,32 @@ mod tests {
let changed_info = put_real_cold_fill_object(&store, &bucket, object, &changed_body).await;
assert_ne!(initial_info.etag, changed_info.etag);
if conditional {
let (paths, _) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await;
for path in paths {
tokio::fs::remove_dir_all(
path.join(&bucket)
.join(object)
.join(changed_info.data_dir.expect("replacement data directory").to_string()),
)
.await
.expect("remove replacement shards before fallback");
}
}
producer.relinquish_or_finish(ColdFillError::Storage(StorageError::Timeout));
let mut response = tokio::time::timeout(Duration::from_secs(10), request)
let result = tokio::time::timeout(Duration::from_secs(10), request)
.await
.expect("generation-changing GET must complete")
.expect("generation-changing GET task must join")
.expect("generation-changing GET must fall back successfully");
.expect("generation-changing GET task must join");
if conditional {
let error = result.expect_err("fallback must recheck the replacement ETag before body access");
assert_eq!(error.code(), &S3ErrorCode::NotModified);
assert_eq!(coordinator.global_waiter_count_for_test(), 0);
assert_eq!(coordinator.active_session_count_for_test(), 0);
return;
}
let mut response = result.expect("generation-changing GET must fall back successfully");
let mut response_body = response.output.body.take().expect("GET response must include a body");
let mut actual = Vec::with_capacity(changed_body.len());
while let Some(chunk) = response_body.next().await {
@@ -10977,53 +11470,6 @@ mod tests {
assert_eq!(DefaultObjectUsecase::get_object_parts_count(Some(1), &single_part_multipart), Some(1));
}
#[test]
fn cold_fill_conditions_fail_before_phase_probe_advances() {
fn run_phase_probe(headers: &HeaderMap, info: &ObjectInfo) -> (S3Result<()>, [usize; 3]) {
let coordination = AtomicUsize::new(0);
let permit = AtomicUsize::new(0);
let reader = AtomicUsize::new(0);
let result = DefaultObjectUsecase::validate_get_object_before_cold_fill(headers, None, info);
if result.is_ok() {
coordination.fetch_add(1, AtomicOrdering::Relaxed);
permit.fetch_add(1, AtomicOrdering::Relaxed);
reader.fetch_add(1, AtomicOrdering::Relaxed);
}
(
result,
[
coordination.load(AtomicOrdering::Relaxed),
permit.load(AtomicOrdering::Relaxed),
reader.load(AtomicOrdering::Relaxed),
],
)
}
let info = ObjectInfo {
etag: Some("phase-etag".to_string()),
parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
number: 1,
..Default::default()
}]),
..Default::default()
};
let mut not_modified = HeaderMap::new();
not_modified.insert(http::header::IF_NONE_MATCH, HeaderValue::from_static("\"phase-etag\""));
let (result, phases) = run_phase_probe(&not_modified, &info);
assert_eq!(result.expect_err("matching If-None-Match must reject").code(), &S3ErrorCode::NotModified);
assert_eq!(phases, [0, 0, 0]);
let mut precondition_failed = HeaderMap::new();
precondition_failed.insert(http::header::IF_MATCH, HeaderValue::from_static("\"other-etag\""));
let (result, phases) = run_phase_probe(&precondition_failed, &info);
assert_eq!(
result.expect_err("mismatched If-Match must reject").code(),
&S3ErrorCode::PreconditionFailed
);
assert_eq!(phases, [0, 0, 0]);
}
#[tokio::test]
async fn execute_get_object_rejects_range_with_part_number() {
let input = GetObjectInput::builder()
+9 -39
View File
@@ -312,10 +312,6 @@ impl DefaultObjectUsecase {
version_id,
part_number,
range,
if_none_match,
if_match,
if_modified_since,
if_unmodified_since,
..
} = req.input.clone();
@@ -343,6 +339,7 @@ impl DefaultObjectUsecase {
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
.await
.map_err(ApiError::from)?;
let opts = internal_object_info_lookup_opts(opts);
prepare_odm_read_generation(&store, &mut req, &bucket).await;
@@ -413,41 +410,6 @@ impl DefaultObjectUsecase {
opts.version_id.as_deref(),
));
}
if let Some(match_etag) = if_none_match
&& let Some(strong_etag) = match_etag.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) == strong_etag)
{
return Err(S3Error::new(S3ErrorCode::NotModified));
}
if let Some(modified_since) = if_modified_since {
// obj_time < givenTime + 1s
if info.mod_time.is_some_and(|mod_time| {
let give_time: OffsetDateTime = modified_since.into();
mod_time < give_time.add(time::Duration::seconds(1))
}) {
return Err(S3Error::new(S3ErrorCode::NotModified));
}
}
if let Some(match_etag) = if_match {
if let Some(strong_etag) = match_etag.into_etag()
&& info
.etag
.as_ref()
.is_some_and(|etag| ETag::Strong(etag.clone()) != strong_etag)
{
return Err(S3Error::new(S3ErrorCode::PreconditionFailed));
}
} else if let Some(unmodified_since) = if_unmodified_since
&& info.mod_time.is_some_and(|mod_time| {
let give_time: OffsetDateTime = unmodified_since.into();
mod_time > give_time.add(time::Duration::seconds(1))
})
{
return Err(S3Error::new(S3ErrorCode::PreconditionFailed));
}
// An authorized replication convergence check only needs etag/size/mtime
// to compare source and replica; it holds no customer key, so the SSE-C
// read validation is skipped for it (and only it).
@@ -465,6 +427,14 @@ impl DefaultObjectUsecase {
)?;
}
let conditions = check_preconditions(&req.headers, &info);
if conditions.is_err() && !replication_check {
let principal = SseKmsPrincipal::from_request(&req);
crate::storage::sse::validate_conditional_read(&bucket, &key, &info.user_defined, &req.headers, principal.as_ref())
.await?;
}
conditions?;
// Compute x-amz-expiration header from lifecycle prediction (before info is partially moved)
let expiration_header = resolve_put_object_expiration(&bucket, &info).await;
// Clone ObjectInfo for event notification only when an event will
+2 -1
View File
@@ -205,6 +205,8 @@ fn remove_source_replication_bookkeeping(user_defined: &mut HashMap<String, Stri
}
}
#[cfg(test)]
mod conditional_read_bench;
mod copy;
mod delete;
mod extract;
@@ -238,7 +240,6 @@ use self::test_support::*;
use std::collections::HashMap;
use std::io;
use std::ops::Add;
use std::path::Path;
use std::pin::Pin;
use std::task::{Context, Poll};
+8 -20
View File
@@ -1092,30 +1092,18 @@ pub(crate) fn odm_check_source_preconditions(headers: &HeaderMap, head: &SourceH
if (needs_etag && head.etag.is_none()) || (needs_mtime && head.last_modified.is_none()) {
return Err(odm_source_unavailable_error("missing_source_validator"));
}
let mut metadata = HashMap::new();
for (name, value) in [("cache-control", &head.cache_control), ("expires", &head.expires)] {
if let Some(value) = value {
metadata.insert(name.to_owned(), value.clone());
}
}
let info = ObjectInfo {
etag: head.etag.clone(),
mod_time: head.last_modified.map(OffsetDateTime::from),
user_defined: metadata.into(),
..Default::default()
};
// A successful source read establishes wildcard existence, but the
// remaining conditions must still run in their ordinary precedence.
if head.etag.is_none() && (if_match == Some("*") || if_none_match == Some("*")) {
let mut remaining = headers.clone();
if if_match == Some("*") {
remaining.remove(http::header::IF_MATCH);
remaining.remove(http::header::IF_UNMODIFIED_SINCE);
}
if if_none_match == Some("*") {
remaining.remove(http::header::IF_NONE_MATCH);
remaining.remove(http::header::IF_MODIFIED_SINCE);
}
check_preconditions(&remaining, &info)?;
return if if_none_match == Some("*") {
Err(S3Error::new(S3ErrorCode::NotModified))
} else {
Ok(())
};
}
check_preconditions(headers, &info)
}
@@ -1513,7 +1501,7 @@ mod tests {
#[test]
fn build_put_like_object_lock_metadata_rejects_retain_until_date_without_mode() {
let retain_until = Timestamp::from(OffsetDateTime::now_utc().add(time::Duration::days(1)));
let retain_until = Timestamp::from(OffsetDateTime::now_utc() + time::Duration::days(1));
let err = build_put_like_object_lock_metadata(
"test-bucket",
&metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+5 -3
View File
@@ -1190,8 +1190,10 @@ pub(crate) mod object_usecase {
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_object::lookup_get_object_body_cache_hook;
pub(crate) use crate::storage::storage_api::ecstore_object::{
GetObjectBodyCacheHookLookup, get_object_body_cache_plaintext_len,
GetObjectBodyCacheHookLookup, GetObjectReadCondition, GetObjectReadMetadata, get_object_body_cache_plaintext_len,
with_get_object_read_condition,
};
pub(crate) use crate::storage::storage_api::{evaluate_read_preconditions, has_read_preconditions};
}
pub(crate) mod contract {
@@ -1322,8 +1324,8 @@ pub(crate) mod test {
};
pub(crate) mod set_disk {
pub(crate) use crate::storage::storage_api::ecstore_set_disk::{
MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause,
fail_next_quota_ledger_save_for_test,
ConditionalReadBenchmarkMetadataGuard, MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier,
PutObjectCommitPause, fail_next_quota_ledger_save_for_test,
};
}
+285
View File
@@ -3290,6 +3290,291 @@ mod tests {
assert_eq!(bytes_polled.load(Ordering::Relaxed), 0);
}
mod conditional_object_http {
use super::*;
use crate::app::storage_api::object_usecase::{StorageObjectInfo, check_preconditions};
use crate::server::hybrid::HybridBody;
#[derive(Clone)]
struct ConditionalObjectS3 {
calls: Arc<AtomicUsize>,
}
#[derive(Debug)]
struct FixtureTransportError(String);
impl std::fmt::Display for FixtureTransportError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for FixtureTransportError {}
type HttpConditionCase<'a> = (&'a str, &'a [(&'a str, &'a str)], StatusCode, bool);
impl<'a> From<Box<dyn std::error::Error + Send + Sync + 'a>> for FixtureTransportError {
fn from(error: Box<dyn std::error::Error + Send + Sync + 'a>) -> Self {
Self(error.to_string())
}
}
fn object_info() -> StorageObjectInfo {
StorageObjectInfo {
etag: Some("abc".to_owned()),
mod_time: Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixture timestamp")),
user_defined: HashMap::from([("cache-control".to_owned(), "max-age=60".to_owned())]).into(),
..Default::default()
}
}
#[async_trait::async_trait]
impl s3s::S3 for ConditionalObjectS3 {
async fn get_object(
&self,
req: s3s::S3Request<s3s::dto::GetObjectInput>,
) -> s3s::S3Result<s3s::S3Response<s3s::dto::GetObjectOutput>> {
self.calls.fetch_add(1, Ordering::Relaxed);
assert_eq!(req.input.bucket, "bucket");
assert_eq!(req.input.key, "object");
assert_eq!(req.input.if_match.is_some(), req.headers.contains_key(http::header::IF_MATCH));
assert_eq!(req.input.if_none_match.is_some(), req.headers.contains_key(http::header::IF_NONE_MATCH));
check_preconditions(&req.headers, &object_info())?;
Ok(s3s::S3Response::new(s3s::dto::GetObjectOutput {
body: Some(s3s::dto::StreamingBlob::from_bytes(Bytes::from_static(b"payload"))),
content_length: Some(7),
e_tag: Some(s3s::dto::ETag::Strong("abc".to_owned())),
..Default::default()
}))
}
async fn head_object(
&self,
req: s3s::S3Request<s3s::dto::HeadObjectInput>,
) -> s3s::S3Result<s3s::S3Response<s3s::dto::HeadObjectOutput>> {
self.calls.fetch_add(1, Ordering::Relaxed);
assert_eq!(req.input.bucket, "bucket");
assert_eq!(req.input.key, "object");
assert_eq!(req.input.if_match.is_some(), req.headers.contains_key(http::header::IF_MATCH));
assert_eq!(req.input.if_none_match.is_some(), req.headers.contains_key(http::header::IF_NONE_MATCH));
check_preconditions(&req.headers, &object_info())?;
Ok(s3s::S3Response::new(s3s::dto::HeadObjectOutput {
content_length: Some(7),
e_tag: Some(s3s::dto::ETag::Strong("abc".to_owned())),
..Default::default()
}))
}
}
#[tokio::test]
async fn conditional_get_head_http1_http2_parse_headers_and_apply_precedence() {
let cases: &[HttpConditionCase<'_>] = &[
("strong list", &[("if-none-match", "\"other\", \"abc\"")], StatusCode::NOT_MODIFIED, true),
("weak list", &[("if-none-match", "\"other\", W/\"abc\"")], StatusCode::NOT_MODIFIED, true),
("weak tag", &[("if-none-match", "W/\"abc\"")], StatusCode::NOT_MODIFIED, true),
("wildcard", &[("if-none-match", "*")], StatusCode::NOT_MODIFIED, true),
("strong match list", &[("if-match", "\"other\", \"abc\"")], StatusCode::OK, true),
("weak match fails", &[("if-match", "W/\"abc\"")], StatusCode::PRECONDITION_FAILED, true),
("wildcard match", &[("if-match", "*")], StatusCode::OK, true),
(
"If-Match precedes matching If-None-Match",
&[("if-match", "\"other\""), ("if-none-match", "\"abc\"")],
StatusCode::PRECONDITION_FAILED,
true,
),
(
"matching If-Match does not suppress If-None-Match",
&[("if-match", "\"abc\""), ("if-none-match", "W/\"abc\"")],
StatusCode::NOT_MODIFIED,
true,
),
(
"If-Match suppresses If-Unmodified-Since",
&[
("if-match", "\"abc\""),
("if-unmodified-since", "Tue, 14 Nov 2023 22:13:19 GMT"),
],
StatusCode::OK,
true,
),
(
"If-None-Match suppresses If-Modified-Since",
&[
("if-none-match", "\"other\""),
("if-modified-since", "Tue, 14 Nov 2023 22:13:21 GMT"),
],
StatusCode::OK,
true,
),
(
"If-Unmodified-Since precedes wildcard If-None-Match",
&[
("if-unmodified-since", "Tue, 14 Nov 2023 22:13:19 GMT"),
("if-none-match", "*"),
],
StatusCode::PRECONDITION_FAILED,
true,
),
(
"date matches at same second",
&[("if-modified-since", "Tue, 14 Nov 2023 22:13:20 GMT")],
StatusCode::NOT_MODIFIED,
true,
),
("unmatched list", &[("if-none-match", "\"other\", W/\"different\"")], StatusCode::OK, true),
(
"empty list member is allowed",
&[("if-none-match", "\"other\", , \"abc\"")],
StatusCode::NOT_MODIFIED,
true,
),
// s3s accepts these quoted fields; the canonical matcher must validate every member.
(
"adjacent tags cannot validate cache",
&[("if-none-match", "\"other\"\"abc\"")],
StatusCode::OK,
true,
),
(
"adjacent tags cannot satisfy If-Match",
&[("if-match", "\"other\"\"abc\"")],
StatusCode::PRECONDITION_FAILED,
true,
),
(
"matching prefix with trailing garbage",
&[("if-none-match", "\"abc\"garbage\"")],
StatusCode::OK,
true,
),
(
"matching prefix cannot satisfy If-Match",
&[("if-match", "\"abc\"garbage\"")],
StatusCode::PRECONDITION_FAILED,
true,
),
(
"matching first tag with malformed later member",
&[("if-none-match", "\"abc\", \"other\"\"different\"")],
StatusCode::OK,
true,
),
(
"If-Match validates malformed later member",
&[("if-match", "\"abc\", \"other\"\"different\"")],
StatusCode::PRECONDITION_FAILED,
true,
),
// These existing s3s parse boundaries reject the request before the handler.
("malformed ETag", &[("if-none-match", "\"abc")], StatusCode::BAD_REQUEST, false),
("invalid date", &[("if-modified-since", "not-a-date")], StatusCode::BAD_REQUEST, false),
(
"obsolete date",
&[("if-modified-since", "Tuesday, 14-Nov-23 22:13:20 GMT")],
StatusCode::BAD_REQUEST,
false,
),
(
"duplicate condition",
&[("if-none-match", "\"other\""), ("if-none-match", "\"abc\"")],
StatusCode::BAD_REQUEST,
false,
),
];
for http2 in [false, true] {
for method in [Method::GET, Method::HEAD] {
for (name, headers, expected_status, reaches_handler) in cases {
let calls = Arc::new(AtomicUsize::new(0));
let inner = S3ServiceBuilder::new(ConditionalObjectS3 {
calls: Arc::clone(&calls),
})
.build();
let service = ServiceBuilder::new()
.layer(BodylessStatusFixLayer)
.layer(HeadRequestBodyFixLayer)
.map_response(|response: Response<s3s::Body>| {
response.map(|rest_body| HybridBody::<s3s::Body, Empty<Bytes>>::Rest { rest_body })
})
.map_err(|error: s3s::HttpError| FixtureTransportError(error.to_string()))
.service(inner);
let (client_io, server_io) = tokio::io::duplex(64 * 1024);
let server = tokio::spawn(async move {
let service = TowerToHyperService::new(service);
if http2 {
hyper::server::conn::http2::Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(server_io), service)
.await
.expect("HTTP/2 server");
} else {
hyper::server::conn::http1::Builder::new()
.serve_connection(TokioIo::new(server_io), service)
.await
.expect("HTTP/1 server");
}
});
let mut request = HttpRequest::builder()
.method(method.clone())
.uri("http://localhost/bucket/object");
for (header, value) in *headers {
request = request.header(*header, *value);
}
let request = request.body(Empty::<Bytes>::new()).expect("conditional request");
let (response, client) = if http2 {
let (mut sender, connection) =
hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(client_io))
.await
.expect("HTTP/2 handshake");
let client = tokio::spawn(async move {
connection.await.expect("HTTP/2 client");
});
let response = tokio::time::timeout(Duration::from_secs(5), sender.send_request(request))
.await
.expect("HTTP/2 response deadline")
.expect("HTTP/2 response");
(response, client)
} else {
let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(client_io))
.await
.expect("HTTP/1 handshake");
let client = tokio::spawn(async move {
connection.await.expect("HTTP/1 client");
});
let response = tokio::time::timeout(Duration::from_secs(5), sender.send_request(request))
.await
.expect("HTTP/1 response deadline")
.expect("HTTP/1 response");
(response, client)
};
assert_eq!(response.status(), *expected_status, "{name}: {method}, HTTP/2={http2}");
assert_eq!(calls.load(Ordering::Relaxed), usize::from(*reaches_handler), "{name}: parser routing");
if *expected_status == StatusCode::NOT_MODIFIED {
assert_eq!(response.headers()[http::header::ETAG], "\"abc\"", "{name}");
assert_eq!(response.headers()[http::header::CACHE_CONTROL], "max-age=60", "{name}");
assert!(response.headers().contains_key(http::header::LAST_MODIFIED), "{name}");
assert!(response.headers().contains_key(http::header::DATE), "{name}");
assert!(!response.headers().contains_key(CONTENT_LENGTH), "{name}");
assert!(!response.headers().contains_key(http::header::TRANSFER_ENCODING), "{name}");
}
let bytes = tokio::time::timeout(Duration::from_secs(5), response.into_body().collect())
.await
.expect("body deadline")
.expect("response body")
.to_bytes();
if method == Method::HEAD || *expected_status == StatusCode::NOT_MODIFIED {
assert!(bytes.is_empty(), "{name}: {method}, HTTP/2={http2}");
} else if *expected_status == StatusCode::OK {
assert_eq!(&bytes[..], b"payload", "{name}");
} else if *expected_status == StatusCode::PRECONDITION_FAILED {
assert!(String::from_utf8_lossy(&bytes).contains("<Code>PreconditionFailed</Code>"), "{name}");
}
client.abort();
server.abort();
}
}
}
}
}
#[derive(Clone, Copy)]
struct UploadPartTimeoutS3 {
timeout: Duration,
+59 -9
View File
@@ -1175,15 +1175,9 @@ where
/// responses whose HTTP status code MUST NOT carry a body per RFC 9110 §6.4.1
/// and §15 (1xx, 204, 205, 304).
///
/// The inner s3s layer serializes every `S3Error` — including 304 `NotModified`
/// preconditions — as an XML body. Returning that body for a 304 is a protocol
/// violation: hyper's HTTP/1.1 encoder forces the body to zero length but
/// preserves the response, while the HTTP/2 path fills in `content-length`
/// from the body's size hint and writes DATA frames after a HEADERS frame that
/// should have carried END_STREAM. h2 clients (curl, browsers) and proxies see
/// the malformed response as a connection-level failure — in the wild this
/// surfaces as `GOAWAY error=0` on h2 and as an upstream-disconnect 5xx from
/// reverse proxies like ngrok (`ERR_NGROK_3004`).
/// s3s already suppresses bodyless error responses. This layer also protects
/// REST routes and custom responses: unlike the HTTP/1.1 encoder, hyper's
/// HTTP/2 sender relies on the body rather than the status to end the stream.
#[derive(Clone)]
pub struct BodylessStatusFixLayer;
@@ -5312,6 +5306,62 @@ mod tests {
assert!(bytes.is_empty(), "304 response body must be empty");
}
#[tokio::test]
async fn bodyless_304_is_valid_on_http1_and_http2() {
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::service::TowerToHyperService;
for http2 in [false, true] {
let (client_io, server_io) = tokio::io::duplex(4096);
let service = TowerToHyperService::new(BodylessStatusFixLayer.layer(FixedResponse {
status: StatusCode::NOT_MODIFIED,
body: Bytes::from_static(b"<Error><Code>NotModified</Code></Error>"),
content_type: Some("application/xml"),
}));
let server = tokio::spawn(async move {
if http2 {
hyper::server::conn::http2::Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(server_io), service)
.await
.expect("HTTP/2 server");
} else {
hyper::server::conn::http1::Builder::new()
.serve_connection(TokioIo::new(server_io), service)
.await
.expect("HTTP/1 server");
}
});
let request = Request::builder()
.uri("http://localhost/object")
.body(Empty::<Bytes>::new())
.expect("request");
let (response, client) = if http2 {
let (mut sender, connection) =
hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(client_io))
.await
.expect("HTTP/2 handshake");
let client = tokio::spawn(async move {
connection.await.expect("HTTP/2 client");
});
(sender.send_request(request).await.expect("HTTP/2 response"), client)
} else {
let (mut sender, connection) = hyper::client::conn::http1::handshake(TokioIo::new(client_io))
.await
.expect("HTTP/1 handshake");
let client = tokio::spawn(async move {
connection.await.expect("HTTP/1 client");
});
(sender.send_request(request).await.expect("HTTP/1 response"), client)
};
assert_eq!(response.status(), StatusCode::NOT_MODIFIED);
assert_eq!(response.headers()[http::header::ETAG], "\"abc123\"");
assert!(!response.headers().contains_key(http::header::CONTENT_LENGTH));
assert!(!response.headers().contains_key(http::header::CONTENT_TYPE));
assert!(collect_body(response.into_body()).await.is_empty());
client.abort();
server.abort();
}
}
#[tokio::test]
async fn strips_body_for_204() {
let mut svc = BodylessStatusFixLayer.layer(FixedResponse {
+106 -76
View File
@@ -37,7 +37,6 @@ use s3s::dto::{
use s3s::{S3Error, S3ErrorCode, S3Response, S3Result};
use serde_urlencoded::from_bytes;
use std::collections::HashMap;
use std::ops::Add;
use std::sync::{Arc, OnceLock, RwLock};
use std::time::{Duration, Instant};
use time::OffsetDateTime;
@@ -529,22 +528,29 @@ pub(crate) fn validate_bucket_object_lock_enabled_state(bucket: &str, state: &Ob
/// has not changed), allowing the caller to translate this into the correct
/// HTTP response.
pub(crate) fn check_preconditions(headers: &HeaderMap, info: &ObjectInfo) -> S3Result<()> {
let mod_time = info.mod_time;
let etag = info.etag.as_deref();
match evaluate_read_preconditions(headers, info.etag.as_deref(), info.mod_time) {
Ok(()) => Ok(()),
Err(S3ErrorCode::NotModified) => Err(not_modified_error(info)?),
Err(code) => Err(S3Error::new(code)),
}
}
pub(crate) fn evaluate_read_preconditions(
headers: &HeaderMap,
etag: Option<&str>,
mod_time: Option<OffsetDateTime>,
) -> Result<(), S3ErrorCode> {
let if_match = non_empty_header_value(headers, IF_MATCH);
let if_none_match = non_empty_header_value(headers, IF_NONE_MATCH);
let if_modified_since = non_empty_header_value(headers, IF_MODIFIED_SINCE);
let if_unmodified_since = non_empty_header_value(headers, IF_UNMODIFIED_SINCE);
if mod_time.is_none() && etag.is_none() {
return Ok(());
}
// If-Match: requires ETag to exist
// The caller has resolved an existing object, so '*' does not require an ETag.
if let Some(if_match_val) = if_match {
match etag {
Some(e) if is_etag_equal(e, if_match_val) => {}
_ => return Err(S3Error::new(S3ErrorCode::PreconditionFailed)),
_ if if_match_val == "*" => {}
Some(e) if read_etag_matches(e, if_match_val, false) => {}
_ => return Err(S3ErrorCode::PreconditionFailed),
}
}
@@ -553,32 +559,16 @@ pub(crate) fn check_preconditions(headers: &HeaderMap, info: &ObjectInfo) -> S3R
&& let Some(t) = mod_time
&& let Some(if_unmodified_since) = if_unmodified_since
&& let Ok(given_time) = time::PrimitiveDateTime::parse(if_unmodified_since, &RFC1123).map(|dt| dt.assume_utc())
&& t > given_time.add(time::Duration::seconds(1))
&& t.unix_timestamp() > given_time.unix_timestamp()
{
return Err(S3Error::new(S3ErrorCode::PreconditionFailed));
return Err(S3ErrorCode::PreconditionFailed);
}
// If-None-Match
if let Some(if_none_match) = if_none_match
&& let Some(e) = etag
&& is_etag_equal(e, if_none_match)
&& (if_none_match == "*" || etag.is_some_and(|e| read_etag_matches(e, if_none_match, true)))
{
let mut error_headers = HeaderMap::new();
if let Ok(etag_header) = parse_etag(e) {
error_headers.insert("etag", etag_header);
}
if let Some(t) = mod_time
&& let Ok(last_modified_str) = t.format(&RFC1123)
&& let Ok(last_modified_header) = HeaderValue::from_str(&last_modified_str)
{
error_headers.insert("last-modified", last_modified_header);
}
let mut s3_error = S3Error::new(S3ErrorCode::NotModified);
s3_error.set_message("Not Modified".to_string());
s3_error.set_status_code(StatusCode::NOT_MODIFIED);
s3_error.set_headers(error_headers);
return Err(s3_error);
return Err(S3ErrorCode::NotModified);
}
// If-Modified-Since (only when If-None-Match is absent — semantics per RFC 7232; dates use RFC 1123 format)
@@ -586,31 +576,94 @@ pub(crate) fn check_preconditions(headers: &HeaderMap, info: &ObjectInfo) -> S3R
&& let Some(t) = mod_time
&& let Some(if_modified_since) = if_modified_since
&& let Ok(given_time) = time::PrimitiveDateTime::parse(if_modified_since, &RFC1123).map(|dt| dt.assume_utc())
&& t < given_time.add(time::Duration::seconds(1))
&& t.unix_timestamp() <= given_time.unix_timestamp()
{
let mut error_headers = HeaderMap::new();
if let Some(e) = etag
&& let Ok(etag_header) = parse_etag(e)
{
error_headers.insert("etag", etag_header);
}
if let Ok(last_modified_str) = t.format(&RFC1123)
&& let Ok(last_modified_header) = HeaderValue::from_str(&last_modified_str)
{
error_headers.insert("last-modified", last_modified_header);
}
let mut s3_error = S3Error::new(S3ErrorCode::NotModified);
s3_error.set_message("Not Modified".to_string());
s3_error.set_status_code(StatusCode::NOT_MODIFIED);
s3_error.set_headers(error_headers);
return Err(s3_error);
return Err(S3ErrorCode::NotModified);
}
Ok(())
}
fn not_modified_error(info: &ObjectInfo) -> S3Result<S3Error> {
let mut headers = HeaderMap::new();
if let Some(etag) = info.etag.as_deref() {
headers.insert(http::header::ETAG, parse_etag(etag).map_err(S3Error::internal_error)?);
}
if let Some(mod_time) = info.mod_time {
let value = mod_time.format(&RFC1123).map_err(S3Error::internal_error)?;
headers.insert(
http::header::LAST_MODIFIED,
HeaderValue::from_str(&value).map_err(S3Error::internal_error)?,
);
}
for name in ["cache-control", "content-location", "vary", "expires"] {
if let Some(value) = info.user_defined.get(name) {
headers.insert(name, HeaderValue::from_str(value).map_err(S3Error::internal_error)?);
}
}
if let Some(expires) = info.expires {
let value = expires.format(&RFC1123).map_err(S3Error::internal_error)?;
headers.insert(http::header::EXPIRES, HeaderValue::from_str(&value).map_err(S3Error::internal_error)?);
}
if let Some(version) = info.version_id {
let value = if version.is_nil() {
"null".to_owned()
} else {
version.to_string()
};
headers.insert("x-amz-version-id", HeaderValue::from_str(&value).map_err(S3Error::internal_error)?);
}
let mut error = S3Error::new(S3ErrorCode::NotModified);
error.set_status_code(StatusCode::NOT_MODIFIED);
error.set_message("Not Modified".to_owned());
error.set_headers(headers);
Ok(error)
}
/// GET/HEAD use weak comparison for If-None-Match and strong comparison for If-Match.
fn read_etag_matches(object_etag: &str, condition: &str, allow_weak: bool) -> bool {
let object_etag = object_etag.trim();
let object_weak = object_etag.starts_with("W/");
let object_value = object_etag.strip_prefix("W/").unwrap_or(object_etag).trim_matches('"');
let mut remaining = condition.trim();
let mut matched = false;
while !remaining.is_empty() {
// Recipients ignore empty list members, but tags still need commas.
if let Some(rest) = remaining.strip_prefix(',') {
remaining = rest.trim();
continue;
}
let weak = remaining.starts_with("W/");
let value = remaining.strip_prefix("W/").unwrap_or(remaining);
let (value, rest) = if let Some(quoted) = value.strip_prefix('"') {
let Some(pair) = quoted.split_once('"') else { return false };
pair
} else {
// Preserve bare tags used by legacy internal callers.
if weak {
return false;
}
let (value, rest) = value.split_at(value.find(',').unwrap_or(value.len()));
let value = value.trim();
if value.is_empty() || value == "*" {
return false;
}
(value, rest)
};
if !value.bytes().all(|byte| byte == b'!' || (byte >= b'#' && byte != 0x7f)) {
return false;
}
matched |= value == object_value && (allow_weak || (!weak && !object_weak));
let rest = rest.trim();
if rest.is_empty() {
return matched;
}
let Some(rest) = rest.strip_prefix(',') else { return false };
remaining = rest.trim();
}
matched
}
fn non_empty_header_value(headers: &HeaderMap, name: http::header::HeaderName) -> Option<&str> {
headers
.get(name)
@@ -619,33 +672,10 @@ fn non_empty_header_value(headers: &HeaderMap, name: http::header::HeaderName) -
.filter(|v| !v.is_empty())
}
/// Compares an object ETag with an ETag value from an HTTP header.
///
/// This helper implements HTTP ETag comparison semantics for headers such as
/// `If-Match` and `If-None-Match`:
/// - Supports the wildcard `*`, which matches any `object_etag`.
/// - Supports comma-separated ETag lists (e.g., `"etag1", "etag2"`), returning
/// `true` if any entry matches `object_etag`.
/// - Automatically trims surrounding whitespace and double quotes from both the
/// header entries and `object_etag` before comparison.
///
/// # Parameters
/// - `object_etag`: The ETag associated with the stored object.
/// - `header_etag`: The raw ETag header value received in the request, which may
/// be a wildcard, a single ETag, or a comma-separated list of ETags.
///
/// # Returns
/// `true` if the header value matches the object ETag according to the above
/// HTTP ETag comparison rules, otherwise `false`.
pub(crate) fn is_etag_equal(object_etag: &str, header_etag: &str) -> bool {
let header_etag = header_etag.trim();
if header_etag == "*" {
return true;
}
header_etag
.split(',')
.map(|s| s.trim().trim_matches('"'))
.any(|e| e == object_etag.trim_matches('"'))
pub(crate) fn has_read_preconditions(headers: &HeaderMap) -> bool {
[IF_MATCH, IF_UNMODIFIED_SINCE, IF_NONE_MATCH, IF_MODIFIED_SINCE]
.into_iter()
.any(|name| non_empty_header_value(headers, name).is_some())
}
/// Converts an object ETag string into an HTTP `HeaderValue` for use in response headers.
+141 -28
View File
@@ -27,11 +27,11 @@ mod tests {
use crate::storage::storage_api::test_consumer::{
BucketMetadata, DEFAULT_READ_BUFFER_SIZE, StorageObjectInfo as ObjectInfo, apply_cors_headers,
apply_default_lock_retention_metadata, bucket_metadata_sys_initialized, check_preconditions, decode_tags_to_map,
get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, get_global_bucket_metadata_sys, is_etag_equal,
matches_origin_pattern, parse_etag, parse_object_lock_legal_hold, parse_object_lock_retention,
process_lambda_configurations, process_queue_configurations, process_topic_configurations,
remove_object_lock_metadata_for_copy, remove_object_lock_retention_metadata, set_bucket_metadata,
validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter,
get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, get_global_bucket_metadata_sys, matches_origin_pattern,
parse_etag, parse_object_lock_legal_hold, parse_object_lock_retention, process_lambda_configurations,
process_queue_configurations, process_topic_configurations, remove_object_lock_metadata_for_copy,
remove_object_lock_retention_metadata, set_bucket_metadata, validate_bucket_object_lock_enabled,
validate_list_object_unordered_with_delimiter,
};
use http::{Extensions, HeaderMap, HeaderValue, Method, StatusCode, Uri};
use rustfs_config::MI_B;
@@ -1191,30 +1191,72 @@ mod tests {
}
#[test]
fn test_is_etag_equal() {
// [1] Header ETag is "*", should return true (match any object ETag)
assert!(is_etag_equal("\"d41d8cd98f00b204e9800998ecf8427e\"", "*"));
fn test_read_etag_comparison() {
for (object_etag, condition, matches) in [
("\"d41d8cd98f00b204e9800998ecf8427e\"", "*", true),
("\"d41d8cd98f00b204e9800998ecf8427e\"", "\"d41d8cd98f00b204e9800998ecf8427e\"", true),
("\"d41d8cd98f00b204e9800998ecf8427e\"", "d41d8cd98f00b204e9800998ecf8427e", true),
("\"12345\"", "\"67890\", \"12345\", \"abcde\"", true),
("\"12345\"", " \"67890\" , \"12345\" , \"abcde\" ", true),
("\"12345\"", "\"67890\"", false),
("\"12345\"", "\"67890\", \"abcde\"", false),
] {
let info = ObjectInfo {
etag: Some(object_etag.to_owned()),
..Default::default()
};
let mut headers = HeaderMap::new();
headers.insert(http::header::IF_NONE_MATCH, HeaderValue::from_str(condition).unwrap());
let result = check_preconditions(&headers, &info);
if matches {
assert_eq!(result.unwrap_err().code(), &S3ErrorCode::NotModified);
} else {
assert!(result.is_ok());
}
}
}
// [2] Exact match (both with double quotes)
assert!(is_etag_equal(
"\"d41d8cd98f00b204e9800998ecf8427e\"",
"\"d41d8cd98f00b204e9800998ecf8427e\""
));
// [3] Exact match (object ETag with quotes, header ETag without)
assert!(is_etag_equal("\"d41d8cd98f00b204e9800998ecf8427e\"", "d41d8cd98f00b204e9800998ecf8427e"));
// [4] Header ETag has multiple values (comma-separated), one matches
assert!(is_etag_equal("\"12345\"", "\"67890\", \"12345\", \"abcde\""));
// [5] Header ETag has multiple values with spaces, one matches after trim
assert!(is_etag_equal("\"12345\"", " \"67890\" , \"12345\" , \"abcde\" "));
// [6] No match (different ETag)
assert!(!is_etag_equal("\"12345\"", "\"67890\""));
// [7] No match in multiple values
assert!(!is_etag_equal("\"12345\"", "\"67890\", \"abcde\""));
#[test]
fn test_read_etag_comparison_requires_complete_field() {
for (etag, condition, matches) in [
("abc", "\"other\"\"abc\"", false),
("abc", "\"abc\"garbage\"", false),
("abc", "\"abc\", \"other\"\"different\"", false),
("abc", "\"abc\", W/ \"other\"", false),
("abc", "\"abc\", \"contains space\"", false),
("abc", "\"abc\", *", false),
("abc", "\"other\", , \"abc\"", true),
("abc", ", \"abc\",,", true),
("a,b", "\"other\", \"a,b\"", true),
("", ",,,", false),
("", "\"\"", true),
] {
let info = ObjectInfo {
etag: Some(etag.to_owned()),
..Default::default()
};
for name in [http::header::IF_MATCH, http::header::IF_NONE_MATCH] {
let mut headers = HeaderMap::new();
headers.insert(name.clone(), HeaderValue::from_str(condition).unwrap());
let result = check_preconditions(&headers, &info);
if name == http::header::IF_MATCH {
if matches {
assert!(result.is_ok(), "{condition}");
} else {
assert_eq!(result.unwrap_err().code(), &S3ErrorCode::PreconditionFailed, "{condition}");
}
} else if matches {
// The empty ETag case validates matching via If-Match above;
// emitting that invalid stored response validator fails closed.
assert!(result.is_err(), "{condition}");
if !etag.is_empty() {
assert_eq!(result.unwrap_err().code(), &S3ErrorCode::NotModified, "{condition}");
}
} else {
assert!(result.is_ok(), "{condition}");
}
}
}
}
#[test]
@@ -1409,6 +1451,77 @@ mod tests {
assert!(check_preconditions(&headers18, &info18).is_ok());
}
#[test]
fn test_check_preconditions_rfc304() {
let modified = OffsetDateTime::from_unix_timestamp(1700000000).expect("valid timestamp");
let mut info = ObjectInfo {
etag: Some("abc".to_owned()),
mod_time: Some(modified),
expires: Some(modified + time::Duration::days(1)),
version_id: Some(uuid::Uuid::nil()),
..Default::default()
};
info.user_defined = std::collections::HashMap::from([("cache-control".to_owned(), "max-age=60".to_owned())]).into();
let mut headers = HeaderMap::new();
headers.insert("if-none-match", HeaderValue::from_static("W/\"abc\""));
let error = check_preconditions(&headers, &info).expect_err("weak ETag must validate the cache");
assert_eq!(error.code(), &S3ErrorCode::NotModified);
let response = error.headers().expect("304 carries cache headers");
assert_eq!(response["etag"], "\"abc\"");
assert_eq!(response["cache-control"], "max-age=60");
assert!(response.contains_key("expires"));
assert!(response.contains_key("last-modified"));
assert_eq!(response["x-amz-version-id"], "null");
headers.insert("if-match", HeaderValue::from_static("\"other\""));
assert_eq!(
check_preconditions(&headers, &info).expect_err("If-Match is first").code(),
&S3ErrorCode::PreconditionFailed
);
headers.remove("if-none-match");
headers.insert("if-match", HeaderValue::from_static("W/\"abc\""));
assert_eq!(
check_preconditions(&headers, &info)
.expect_err("If-Match requires strong comparison")
.code(),
&S3ErrorCode::PreconditionFailed
);
headers.clear();
headers.insert("if-none-match", HeaderValue::from_static("\"other\""));
let date = modified.format(&crate::storage::RFC1123).expect("format timestamp");
headers.insert("if-modified-since", HeaderValue::from_str(&date).expect("valid date header"));
assert!(check_preconditions(&headers, &info).is_ok(), "If-None-Match suppresses the date");
headers.clear();
headers.insert("if-unmodified-since", HeaderValue::from_str(&date).expect("valid date header"));
info.mod_time = Some(modified + time::Duration::seconds(1));
assert_eq!(
check_preconditions(&headers, &info)
.expect_err("next second was modified")
.code(),
&S3ErrorCode::PreconditionFailed
);
headers.clear();
headers.insert("if-none-match", HeaderValue::from_static("*"));
info.etag = None;
assert_eq!(
check_preconditions(&headers, &info)
.expect_err("existing object wildcard needs no ETag")
.code(),
&S3ErrorCode::NotModified
);
headers.clear();
headers.insert("if-match", HeaderValue::from_static("*"));
assert!(check_preconditions(&headers, &info).is_ok());
headers.insert("if-match", HeaderValue::from_static("\"other\", \"abc,def\""));
info.etag = Some("abc,def".to_owned());
assert!(
check_preconditions(&headers, &info).is_ok(),
"commas inside an opaque tag are not list delimiters"
);
}
#[test]
fn test_parse_etag() {
// [1] ETag without quotes → adds quotes
+62
View File
@@ -1391,6 +1391,39 @@ fn stored_envelope_master_key_version(metadata: &HashMap<String, String>) -> Opt
pub(crate) struct SseObjectEncryptionResolver;
/// Checks metadata and key access before a conditional read returns without a body.
pub(crate) async fn validate_conditional_read(
bucket: &str,
object: &str,
metadata: &HashMap<String, String>,
headers: &HeaderMap,
principal: Option<&SseKmsPrincipal>,
) -> Result<(), ApiError> {
let metadata = normalize_encryption_metadata_case(metadata)
.map_err(|error| map_get_object_reader_error(StorageError::Io(std::io::Error::other(error))))?;
validate_sse_headers_for_read(&metadata, headers)?;
let (_, customer_key, customer_key_md5) = extract_ssec_params_from_headers(headers)?;
classify_sse_read_response(DecryptionRequest {
bucket,
key: object,
metadata: &metadata,
sse_customer_key: customer_key.as_ref(),
sse_customer_key_md5: customer_key_md5.as_ref(),
principal,
})
.await?;
SseObjectEncryptionResolver
.resolve_read_material(ReadEncryptionRequest {
bucket,
object,
metadata: &metadata,
headers,
})
.await
.map_err(|error| map_get_object_reader_error(StorageError::Io(std::io::Error::other(error))))?;
Ok(())
}
#[async_trait]
impl ObjectEncryptionResolver for SseObjectEncryptionResolver {
async fn resolve_read_material(
@@ -8588,6 +8621,35 @@ mod tests {
assert_eq!(audit_tag(&tags, "kmsErrorClass").as_deref(), Some("access_denied"));
}
#[tokio::test]
async fn conditional_read_denies_kms_before_unwrap_with_normalized_metadata() {
let (principal, audit) = audited_principal(true, false);
let metadata = sse_kms_object_metadata()
.into_iter()
.map(|(name, value)| (name.to_ascii_uppercase(), value))
.collect();
let error = super::validate_conditional_read("finance", "ledger.csv", &metadata, &HeaderMap::new(), Some(&principal))
.await
.expect_err("conditional read must authorize the real key before unwrap");
assert_eq!(error.code, S3ErrorCode::AccessDenied);
assert_eq!(audit_tag(&audit.audit_tags(), "kmsKeyId").as_deref(), Some("finance-key"));
}
#[cfg(feature = "rio-v2")]
#[tokio::test]
async fn conditional_read_denies_minio_only_kms_before_unwrap() {
let (principal, authorizer) = enforcing_principal(false);
let metadata = HashMap::from([
(super::MINIO_INTERNAL_ENCRYPTION_KMS_SEALED_KEY_HEADER.to_owned(), "sealed-key".to_owned()),
(super::MINIO_INTERNAL_ENCRYPTION_KMS_KEY_ID_HEADER.to_owned(), "finance-key".to_owned()),
]);
let error = super::validate_conditional_read("finance", "ledger.csv", &metadata, &HeaderMap::new(), Some(&principal))
.await
.expect_err("MinIO metadata must not bypass KMS authorization");
assert_eq!(error.code, S3ErrorCode::AccessDenied);
assert_eq!(authorizer.calls(), vec![(KmsAction::DecryptAction, "finance-key".to_owned())]);
}
#[test]
fn read_response_projection_uses_shared_managed_metadata_compatibility() {
let context_json = r#"{"tenant":"s3-select"}"#;
+14 -12
View File
@@ -109,11 +109,12 @@ pub(crate) type StorageSelectObjectSnapshot = ecstore_object::SelectObjectSnapsh
pub(crate) type StorageObjectToDelete = contract::object::ObjectToDelete;
pub(crate) type StoragePutObjReader = super::PutObjReader;
pub(crate) use super::ecfs_extend::{
RFC1123, apply_bucket_default_lock_retention, apply_cors_headers, check_preconditions, get_buffer_size_opt_in,
get_validated_store, load_bucket_object_lock_config_state, parse_object_lock_legal_hold, parse_object_lock_retention,
parse_part_number_i32_to_usize, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
remove_object_lock_metadata_for_copy, validate_bucket_exists, validate_bucket_object_lock_enabled_state,
validate_list_object_unordered_with_delimiter, validate_object_key, wrap_response_with_cors,
RFC1123, apply_bucket_default_lock_retention, apply_cors_headers, check_preconditions, evaluate_read_preconditions,
get_buffer_size_opt_in, get_validated_store, has_read_preconditions, load_bucket_object_lock_config_state,
parse_object_lock_legal_hold, parse_object_lock_retention, parse_part_number_i32_to_usize, process_lambda_configurations,
process_queue_configurations, process_topic_configurations, remove_object_lock_metadata_for_copy, validate_bucket_exists,
validate_bucket_object_lock_enabled_state, validate_list_object_unordered_with_delimiter, validate_object_key,
wrap_response_with_cors,
};
pub(crate) use super::sse::{
DecryptionRequest, EncryptionRequest, ObjectDekRewrapOutcome, PrepareEncryptionRequest, SseKmsPrincipal,
@@ -400,7 +401,7 @@ pub(crate) mod tonic_service_consumer {
pub(crate) mod test_consumer {
pub(crate) use super::super::{
apply_cors_headers, apply_default_lock_retention_metadata, check_preconditions, decode_tags_to_map,
get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, is_etag_equal, matches_origin_pattern, parse_etag,
get_adaptive_buffer_size_with_profile, get_buffer_size_opt_in, matches_origin_pattern, parse_etag,
parse_object_lock_legal_hold, parse_object_lock_retention, process_lambda_configurations, process_queue_configurations,
process_topic_configurations, remove_object_lock_metadata_for_copy, remove_object_lock_retention_metadata,
validate_bucket_object_lock_enabled, validate_list_object_unordered_with_delimiter,
@@ -579,10 +580,11 @@ pub(crate) mod ecstore_object {
pub(crate) use rustfs_ecstore::api::object::GetObjectBodySource;
pub(crate) use rustfs_ecstore::api::object::{
EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup,
ObjectEncryptionResolver, ObjectMutationHook, PrepareSelectObjectSnapshotError, ReadEncryptionMaterial,
ReadEncryptionMode, ReadEncryptionRequest, SelectObjectSnapshot, WriteCommitGuard, WriteCompletion,
get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
GetObjectReadCondition, GetObjectReadMetadata, ObjectEncryptionResolver, ObjectMutationHook,
PrepareSelectObjectSnapshotError, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
SelectObjectSnapshot, WriteCommitGuard, WriteCompletion, get_object_body_cache_plaintext_len,
lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook,
unregister_get_object_body_cache_hook, unregister_object_mutation_hook, with_get_object_read_condition,
};
}
@@ -597,8 +599,8 @@ pub(crate) mod ecstore_test_support {
pub(crate) mod ecstore_set_disk {
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::set_disk::test_util::{
MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier, PutObjectCommitPause,
fail_next_quota_ledger_save_for_test,
ConditionalReadBenchmarkMetadataGuard, MultipartCommitBarrier, MultipartCommitPause, PutObjectCommitBarrier,
PutObjectCommitPause, fail_next_quota_ledger_save_for_test,
};
pub(crate) use rustfs_ecstore::api::set_disk::{
DEFAULT_READ_BUFFER_SIZE, file_info_quorum_hash, get_lock_acquire_timeout, is_valid_storage_class,
+3
View File
@@ -108,6 +108,9 @@ their issue closes.
| `run_object_batch_bench.sh` | dev-tool | Batch object benchmark runner (warp/s3bench) | internode + scanner runbooks |
| `run_object_batch_bench_enhanced.sh` | dev-tool | Enhanced batch benchmark runner; hub used by the smoke rigs | hotpath runbook |
| `run_pinned_paired_abba_bench.sh` | dev-tool | Pinned RustFS/MinIO paired ABBA benchmark orchestrator for backlog#1432 | `test_pinned_paired_abba_bench.sh` |
| `prepare_conditional_read_baseline.py` | dev-tool | Installs only the matching conditional-read benchmark harness in a clean detached baseline | `docs/testing/conditional-read-benchmark.md` |
| `run_conditional_read_abba.py` | dev-tool | Serial ordinary/hit/miss GET ABBA with cache/slow-disk controls and drift/regression gates | `docs/testing/conditional-read-benchmark.md` |
| `test_conditional_read_abba.py` | dev-tool | Parser, drift rejection, and throughput/lock regression tests for conditional-read evidence | `run_conditional_read_abba.py` |
| `run_get_codec_streaming_smoke.sh` | dev-tool | Local GET benchmark harness for the codec streaming read path | `docs/testing/ecstore-validation-suite-design.md` |
| `run_get_1mib_abba_stage_metrics.sh` | dev-tool | Exact-1MiB isolated-host GET ABBA/stage-metrics harness for backlog#1434 | `test_get_1mib_abba_stage_metrics.sh` |
| `issue_2007_coalescer_prometheus_report.py` | dev-tool | Read-only Prometheus report for GET metadata coalescer delay cost validation; usage in the module docstring | `test_issue_2007_coalescer_prometheus_report.sh` |
@@ -0,0 +1,74 @@
#!/usr/bin/env python3
"""Install only the paired benchmark harness in a clean detached baseline."""
import argparse
from pathlib import Path
import re
import subprocess
def replace_once(source, old, new, path):
if source.count(old) != 1:
raise ValueError(f"baseline anchor changed in {path}; inspect it before measuring")
return source.replace(old, new)
def prepare(root, baseline, revision):
if baseline.resolve() == root.resolve():
raise ValueError("baseline must be a separate worktree")
if not re.fullmatch(r"[0-9a-f]{40}", revision):
raise ValueError("supply the full baseline commit SHA")
def git(*args):
return subprocess.check_output(["git", "-C", str(baseline), *args], text=True).strip()
if git("rev-parse", "HEAD") != revision or git("status", "--porcelain"):
raise ValueError("baseline must be clean at the requested commit")
if subprocess.run(["git", "-C", str(baseline), "symbolic-ref", "-q", "HEAD"],
stdout=subprocess.DEVNULL).returncode == 0:
raise ValueError("use a detached baseline worktree")
edits = {}
def edit(path, old, new):
previous = edits.get(path, (baseline / path).read_text())
edits[path] = replace_once(previous, old, new, path)
edit("rustfs/src/app/object/mod.rs", "mod test_support;",
"mod test_support;\n#[cfg(test)]\nmod conditional_read_bench;")
edit("crates/ecstore/src/set_disk/mod.rs",
'const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";',
'#[cfg(feature = "test-util")]\nmod conditional_read_bench;\n'
'#[cfg(feature = "test-util")]\npub use conditional_read_bench::ConditionalReadBenchmarkMetadataGuard;\n\n'
'const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";')
edit("crates/ecstore/src/set_disk/read.rs",
"fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOptions, read_data: bool) -> Option<&'static str> {",
"fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOptions, read_data: bool) -> Option<&'static str> {\n"
' #[cfg(feature = "test-util")]\n if super::conditional_read_bench::cache_bypass_applies(bucket) {\n'
' return Some("benchmark");\n }')
edit("crates/ecstore/src/api/mod.rs",
"pub use crate::bucket::quota::reservation::fail_next_quota_ledger_save_for_test;",
"pub use crate::bucket::quota::reservation::fail_next_quota_ledger_save_for_test;\n"
" pub use crate::set_disk::ConditionalReadBenchmarkMetadataGuard;")
for path in ("rustfs/src/storage/storage_api.rs", "rustfs/src/app/storage_api.rs"):
# The facades also expose these types elsewhere; target the test-util block.
source = (baseline / path).read_text()
marker = ("pub(crate) use rustfs_ecstore::api::set_disk::test_util::{" if "/storage/" in path else
"pub(crate) use crate::storage::storage_api::ecstore_set_disk::{")
replacement = marker + "\n ConditionalReadBenchmarkMetadataGuard,"
edits[path] = replace_once(source, marker, replacement, path)
for path in ("rustfs/src/app/object/conditional_read_bench.rs",
"crates/ecstore/src/set_disk/conditional_read_bench.rs"):
if (baseline / path).exists():
raise ValueError(f"baseline already contains {path}")
edits[path] = (root / path).read_text()
# Resolve every anchor before writing; never apply a partial adapter.
for path, source in edits.items():
(baseline / path).write_text(source)
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--baseline-worktree", type=Path, required=True)
parser.add_argument("--revision", required=True)
args = parser.parse_args()
prepare(Path(__file__).resolve().parent.parent, args.baseline_worktree, args.revision)
if __name__ == "__main__":
main()
+205
View File
@@ -0,0 +1,205 @@
#!/usr/bin/env python3
"""Run the identical application-read test harness in serial ABBA processes.
This measures local EC reads, not network/authentication or deployment capacity.
Use matching toolchain, features and profiles for both executable artifacts.
"""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import shutil
import subprocess
import time
def digest(path):
result = hashlib.sha256()
with path.open("rb") as source:
for chunk in iter(lambda: source.read(1024 * 1024), b""):
result.update(chunk)
return result.hexdigest()
def parse_rows(output):
rows = {}
prefix = "CONDITIONAL_BENCH "
for line in output.splitlines():
if prefix in line:
row = json.loads(line.split(prefix, 1)[1])
key = (row["size"], row["kind"], row["cache"], row["slowtail_ms"])
if key in rows:
raise ValueError(f"duplicate benchmark row {key}")
rows[key] = row
if len(rows) != 6:
raise ValueError(f"expected six benchmark rows, received {len(rows)}")
if {(key[0], key[1]) for key in rows} != {
(size, kind) for size in (4096, 1_300_000) for kind in ("unconditional", "hit", "miss")
}:
raise ValueError("benchmark workload matrix is incomplete")
for row in rows.values():
for metric in ("ops_per_sec", "p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms"):
if not math.isfinite(row[metric]) or row[metric] <= 0:
raise ValueError(f"invalid benchmark metric {metric}")
return rows
def check_resource_isolation(check_load=True):
commands = subprocess.check_output(["ps", "-A", "-o", "comm="], text=True)
if any(Path(command.strip()).name in ("cargo", "rustc") for command in commands.splitlines()):
raise RuntimeError("stop concurrent Cargo builds before measuring")
load = os.getloadavg()
if check_load and load[0] > (os.cpu_count() or 1) * 0.5:
raise RuntimeError(f"host load too high for paired measurements: {load[0]}")
return load
def run_case(executable, expected_digest, env, log_path=None):
if digest(executable) != expected_digest:
raise RuntimeError("benchmark executable changed before execution")
# Poll for concurrent builds without attributing this benchmark's own CPU
# use to external host load. Load gates run before and after the process.
with subprocess.Popen([str(executable.resolve()), "conditional_read_paired_benchmark",
"--ignored", "--nocapture", "--test-threads=1"],
env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) as run:
try:
while True:
try:
output, _ = run.communicate(timeout=1)
break
except subprocess.TimeoutExpired:
check_resource_isolation(check_load=False)
except BaseException:
run.kill()
output, _ = run.communicate()
if log_path is not None:
log_path.write_text(output)
raise
returncode = run.returncode
if log_path is not None:
log_path.write_text(output)
check_resource_isolation()
if digest(executable) != expected_digest:
raise RuntimeError("benchmark executable changed during execution")
return returncode, output
def drift_passes(first, last):
if abs(last["ops_per_sec"] / first["ops_per_sec"] - 1) > 0.10:
return False
return all(
abs(last[metric] - first[metric]) <= max(first[metric] * 0.15, 0.25)
for metric in ("p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")
)
def compare(rows):
if not all(rows[0].keys() == leg.keys() for leg in rows):
raise ValueError("ABBA legs have different workload matrices")
unstable = [key for key in rows[0] if not drift_passes(rows[0][key], rows[3][key])]
if unstable:
return {"accepted": False, "reason": "baseline_drift", "unstable_cases": unstable}
results = []
for key, first in rows[0].items():
baseline = {metric: math.sqrt(first[metric] * rows[3][key][metric]) for metric in
("ops_per_sec", "p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")}
candidate = {metric: math.sqrt(rows[1][key][metric] * rows[2][key][metric]) for metric in baseline}
# Protect ordinary reads and condition misses; 304 savings alone do not pass.
passed = candidate["ops_per_sec"] >= baseline["ops_per_sec"] * 0.95 and all(
candidate[metric] <= baseline[metric] + max(baseline[metric] * 0.10, 0.25)
for metric in ("p95_ms", "p99_ms", "lock_p95_ms", "lock_p99_ms")
)
results.append({"case": key, "baseline": baseline, "candidate": candidate,
"within_regression_budget": passed})
return {"accepted": all(row["within_regression_budget"] for row in results), "cases": results}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--baseline", type=Path, required=True)
parser.add_argument("--candidate", type=Path, required=True)
parser.add_argument("--baseline-revision", required=True)
parser.add_argument("--candidate-revision", required=True)
parser.add_argument("--harness", type=Path, required=True)
parser.add_argument("--profile", required=True)
parser.add_argument("--out-dir", type=Path, required=True)
parser.add_argument("--iterations", type=int, default=1000)
parser.add_argument("--concurrency", type=int, default=8)
parser.add_argument("--cooldown-seconds", type=float, default=5)
args = parser.parse_args()
if args.iterations < 100 or args.concurrency < 1 or args.cooldown_seconds < 0:
parser.error("iterations must be >=100, concurrency >0, cooldown >=0")
args.out_dir.mkdir(parents=True, exist_ok=False)
executables = {}
for name, source in (("baseline", args.baseline), ("candidate", args.candidate)):
original_digest = digest(source)
snapshot = args.out_dir / f"{name}.binary"
shutil.copyfile(source, snapshot)
snapshot.chmod(0o555)
if digest(snapshot) != original_digest:
raise RuntimeError("source executable changed while taking its snapshot")
executables[name] = snapshot
manifest = {
"baseline_revision": args.baseline_revision, "candidate_revision": args.candidate_revision,
"baseline_sha256": digest(executables["baseline"]), "candidate_sha256": digest(executables["candidate"]),
"harness_sha256": digest(args.harness), "profile": args.profile,
"iterations": args.iterations, "concurrency": args.concurrency,
"order": ["A1", "B1", "B2", "A2"], "slowtail_ms": [0, 20], "cache": [False, True],
"drift_budget": "throughput 10%; latency and lock percentiles 15% or 0.25ms",
"regression_budget": "throughput -5%; latency and lock percentiles +10% or 0.25ms",
"cpu_count": os.cpu_count(), "load_at_start": check_resource_isolation(),
}
(args.out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
legs = []
for name, artifact in (("A1", "baseline"), ("B1", "candidate"),
("B2", "candidate"), ("A2", "baseline")):
executable = executables[artifact]
cases = {}
for cache in (False, True):
for delay in (0, 20):
check_resource_isolation()
env = os.environ.copy()
# Keep ambient experiments and proxies out of localhost-sensitive runs.
for variable in tuple(env):
if variable.startswith("RUSTFS_") or variable.lower().endswith("_proxy"):
del env[variable]
env.update({
"NO_PROXY": "*", "no_proxy": "*", "RUSTFS_OBJECT_LOCK_DIAG_ENABLE": "true",
"RUSTFS_OBJECT_DATA_CACHE_ENABLE": str(cache).lower(),
"RUSTFS_OBJECT_DATA_CACHE_MODE": "fill_materialize_enabled",
"RUSTFS_OBJECT_DATA_CACHE_MAX_BYTES": "8388608",
"RUSTFS_OBJECT_DATA_CACHE_MAX_ENTRY_BYTES": "2097152",
"RUSTFS_OBJECT_DATA_CACHE_MIN_FREE_MEMORY_PERCENT": "0",
"RUSTFS_CONDITIONAL_BENCH_ITERATIONS": str(args.iterations),
"RUSTFS_CONDITIONAL_BENCH_CONCURRENCY": str(args.concurrency),
"RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS": str(delay),
"RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS": "3",
"RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET": "conditional-bench",
"RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX": "bench-",
})
returncode, output = run_case(executable, manifest[f"{artifact}_sha256"], env,
args.out_dir / f"{name}-cache{int(cache)}-delay{delay}.log")
if returncode:
raise RuntimeError(f"{name} cache={cache} delay={delay} failed; see saved log")
for key, row in parse_rows(output).items():
if (row["cache"] != cache or row["slowtail_ms"] != str(delay)
or row["iterations"] != args.iterations or row["concurrency"] != args.concurrency
or row["metadata_cache_hits"] != 0 or not row["metadata_cache_bypassed"]
or row["lock_samples"] <= 0):
raise ValueError("benchmark did not observe requested environment")
cases[key] = row
print(f"{name}: cache={cache} slowtail={delay}ms complete", flush=True)
legs.append(cases)
time.sleep(args.cooldown_seconds)
result = compare(legs)
result["load_at_end"] = os.getloadavg()
(args.out_dir / "summary.json").write_text(json.dumps(result, indent=2) + "\n")
print(json.dumps(result, indent=2))
return 0 if result["accepted"] else 2
if __name__ == "__main__":
raise SystemExit(main())
+116
View File
@@ -0,0 +1,116 @@
#!/usr/bin/env python3
"""Verify that paired conditional-read evidence cannot bypass acceptance gates."""
import copy
import json
from pathlib import Path
import subprocess
import unittest
from unittest.mock import MagicMock, patch
from run_conditional_read_abba import compare, parse_rows, run_case
def row():
return {"size": 4096, "kind": "miss", "cache": False, "slowtail_ms": "0",
"ops_per_sec": 1000, "p95_ms": 2, "p99_ms": 3,
"lock_p95_ms": 1, "lock_p99_ms": 2}
class AcceptanceTests(unittest.TestCase):
def test_stable_equivalent_legs_pass(self):
self.assertTrue(compare([{("case",): row()} for _ in range(4)])["accepted"])
def test_baseline_drift_blocks_candidate_attribution(self):
legs = [{("case",): row()} for _ in range(4)]
legs[3][("case",)]["ops_per_sec"] = 800
result = compare(legs)
self.assertFalse(result["accepted"])
self.assertEqual(result["reason"], "baseline_drift")
self.assertNotIn("cases", result)
def test_miss_throughput_and_lock_regressions_fail(self):
for metric, value in [("ops_per_sec", 900), ("lock_p99_ms", 4)]:
legs = [{("case",): row()} for _ in range(4)]
for index in (1, 2):
legs[index][("case",)][metric] = value
self.assertFalse(compare(legs)["accepted"], metric)
def test_missing_workload_is_rejected(self):
legs = [{("case",): row()} for _ in range(4)]
legs[1] = {}
with self.assertRaises(ValueError):
compare(legs)
def test_complete_rows_include_first_libtest_output_line(self):
rows = []
for size in (4096, 1_300_000):
for kind in ("unconditional", "hit", "miss"):
value = copy.copy(row())
value.update(size=size, kind=kind)
rows.append("CONDITIONAL_BENCH " + json.dumps(value))
output = "test app::benchmark ... " + "\n".join(rows)
self.assertEqual(len(parse_rows(output)), 6)
with self.assertRaises(ValueError):
parse_rows(output + "\n" + rows[0])
def test_wrong_shape_or_nonfinite_metrics_are_rejected(self):
rows = []
for size in (4096, 1_300_000):
for kind in ("unconditional", "hit", "miss"):
value = copy.copy(row())
value.update(size=size, kind=kind)
rows.append(value)
for field, value in [("kind", "unknown"), ("p99_ms", float("nan"))]:
changed = copy.deepcopy(rows)
changed[0][field] = value
output = "\n".join("CONDITIONAL_BENCH " + json.dumps(item) for item in changed)
with self.assertRaises(ValueError):
parse_rows(output)
class IsolationTests(unittest.TestCase):
def process(self):
process = MagicMock()
process.__enter__.return_value = process
process.communicate.return_value = ("saved output", None)
process.returncode = 0
return process
def test_changed_executable_is_rejected_before_launch(self):
with patch("run_conditional_read_abba.digest", return_value="changed"), \
patch("run_conditional_read_abba.subprocess.Popen") as launch:
with self.assertRaisesRegex(RuntimeError, "before execution"):
run_case(Path("binary"), "expected", {})
launch.assert_not_called()
def test_changed_executable_is_rejected_after_launch(self):
with patch("run_conditional_read_abba.digest", side_effect=["expected", "changed"]), \
patch("run_conditional_read_abba.check_resource_isolation"), \
patch("run_conditional_read_abba.subprocess.Popen", return_value=self.process()):
with self.assertRaisesRegex(RuntimeError, "during execution"):
run_case(Path("binary"), "expected", {})
def test_final_case_checks_resource_isolation_after_exit(self):
with patch("run_conditional_read_abba.digest", return_value="expected"), \
patch("run_conditional_read_abba.check_resource_isolation", side_effect=RuntimeError("concurrent build")), \
patch("run_conditional_read_abba.subprocess.Popen", return_value=self.process()):
with self.assertRaisesRegex(RuntimeError, "concurrent build"):
run_case(Path("binary"), "expected", {})
def test_concurrent_build_during_execution_kills_case_and_saves_log(self):
process = self.process()
process.communicate.side_effect = [subprocess.TimeoutExpired("binary", 1), ("saved output", None)]
log = MagicMock()
with patch("run_conditional_read_abba.digest", return_value="expected"), \
patch("run_conditional_read_abba.check_resource_isolation", side_effect=RuntimeError("concurrent build")) as check, \
patch("run_conditional_read_abba.subprocess.Popen", return_value=process):
with self.assertRaisesRegex(RuntimeError, "concurrent build"):
run_case(Path("binary"), "expected", {}, log)
check.assert_called_once_with(check_load=False)
process.kill.assert_called_once()
log.write_text.assert_called_once_with("saved output")
if __name__ == "__main__":
unittest.main()