From db277b17a455394f2c4a9e7653c087d8eef90472 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 5 Jul 2026 12:03:22 +0800 Subject: [PATCH] fix: harden GET object performance paths (#4271) * fix: harden GET object performance paths * fix: satisfy GET multipart layer guard * fix: keep v1 list markers S3 compatible * perf: tighten GET direct-memory decision * ci: isolate s3tests from scanner workload * refactor: simplify get object body lifecycle * fix: satisfy get object clippy --- crates/ecstore/src/diagnostics/get.rs | 6 + crates/ecstore/src/set_disk/mod.rs | 344 ++++++++++++-- crates/ecstore/src/set_disk/read.rs | 15 + crates/io-metrics/src/lib.rs | 28 ++ ...ssue824-get-pressure-results-2026-07-05.md | 86 ++++ ...default-get-hot-path-runbook-2026-07-05.md | 283 +++++++++++ rustfs/src/app/object_usecase.rs | 441 +++++++++++++----- rustfs/src/storage/ecfs_extend.rs | 5 +- rustfs/src/storage/s3_api/bucket.rs | 16 + rustfs/src/storage/s3_api/multipart.rs | 5 +- rustfs/src/storage/storage_api.rs | 9 +- scripts/s3-tests/run.sh | 11 + 12 files changed, 1092 insertions(+), 157 deletions(-) create mode 100644 docs/benchmark/issue824-get-pressure-results-2026-07-05.md create mode 100644 docs/benchmark/issue829-default-get-hot-path-runbook-2026-07-05.md diff --git a/crates/ecstore/src/diagnostics/get.rs b/crates/ecstore/src/diagnostics/get.rs index 7c896eed0..6c57016ee 100644 --- a/crates/ecstore/src/diagnostics/get.rs +++ b/crates/ecstore/src/diagnostics/get.rs @@ -29,6 +29,9 @@ pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition"; pub(crate) const GET_OBJECT_PATH_SET_DISK: &str = "set_disk"; pub(crate) const GET_DIRECT_MEMORY_SUBPATH_DISK_DATA_BLOCKS: &str = "disk_data_blocks"; pub(crate) const GET_DIRECT_MEMORY_SUBPATH_INLINE_BUFFERED: &str = "inline_buffered"; +pub(crate) const GET_DIRECT_MEMORY_DECISION_USE: &str = "use"; +pub(crate) const GET_DIRECT_MEMORY_DECISION_FALLBACK: &str = "fallback"; +pub(crate) const GET_DIRECT_MEMORY_REASON_NONE: &str = "none"; pub(crate) const GET_CODEC_STREAMING_DECISION_USE: &str = "use"; pub(crate) const GET_CODEC_STREAMING_DECISION_FALLBACK: &str = "fallback"; pub(crate) const GET_CODEC_STREAMING_REASON_NONE: &str = "none"; @@ -299,6 +302,9 @@ mod tests { assert_eq!(GET_READER_BUFFER_PREFETCH, "prefetch"); assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_LEGACY_ENGINE, "codec_streaming_legacy_engine"); assert_eq!(GET_OBJECT_PATH_CODEC_STREAMING_RUSTFS_ENGINE, "codec_streaming_rustfs_engine"); + assert_eq!(GET_DIRECT_MEMORY_DECISION_USE, "use"); + assert_eq!(GET_DIRECT_MEMORY_DECISION_FALLBACK, "fallback"); + assert_eq!(GET_DIRECT_MEMORY_REASON_NONE, "none"); assert_eq!(GET_CODEC_STREAMING_DECISION_USE, "use"); assert_eq!(GET_CODEC_STREAMING_DECISION_FALLBACK, "fallback"); assert_eq!(GET_CODEC_STREAMING_REASON_NONE, "none"); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 55a5d62bb..5a4e53a26 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -968,6 +968,67 @@ struct GetCodecStreamingGate { prefer_data_blocks_first_reader_setup: bool, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetDirectMemoryDecision { + Use { object_size: usize }, + Fallback(GetDirectMemoryFallbackReason), +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GetDirectMemoryFallbackReason { + Disabled, + ThresholdZero, + Range, + PartNumber, + VersionId, + Versioned, + VersionSuspended, + InclFreeVersions, + SkipFreeVersion, + DataMovement, + RawDataMovementRead, + DeleteMarker, + MetadataOnly, + VersionOnly, + Encrypted, + Compressed, + Remote, + ObjectInfoMultipart, + FileInfoMultipart, + InvalidSize, + SizeMismatch, + AboveThreshold, +} + +impl GetDirectMemoryFallbackReason { + const fn as_str(self) -> &'static str { + match self { + Self::Disabled => "disabled", + Self::ThresholdZero => "threshold_zero", + Self::Range => "range", + Self::PartNumber => "part_number", + Self::VersionId => "version_id", + Self::Versioned => "versioned", + Self::VersionSuspended => "version_suspended", + Self::InclFreeVersions => "incl_free_versions", + Self::SkipFreeVersion => "skip_free_version", + Self::DataMovement => "data_movement", + Self::RawDataMovementRead => "raw_data_movement_read", + Self::DeleteMarker => "delete_marker", + Self::MetadataOnly => "metadata_only", + Self::VersionOnly => "version_only", + Self::Encrypted => "encrypted", + Self::Compressed => "compressed", + Self::Remote => "remote", + Self::ObjectInfoMultipart => "object_info_multipart", + Self::FileInfoMultipart => "file_info_multipart", + Self::InvalidSize => "invalid_size", + Self::SizeMismatch => "size_mismatch", + Self::AboveThreshold => "above_threshold", + } + } +} + fn record_get_codec_streaming_gate_decision( object_class: GetCodecStreamingObjectClass, decision: GetCodecStreamingDecision, @@ -987,6 +1048,23 @@ fn record_get_codec_streaming_gate_decision( rustfs_io_metrics::record_get_object_codec_streaming_decision_by_size(outcome, object_class, reason, size_bucket); } +fn record_get_direct_memory_decision( + object_class: GetCodecStreamingObjectClass, + decision: GetDirectMemoryDecision, + size_bucket: &'static str, +) { + let (outcome, reason) = match decision { + GetDirectMemoryDecision::Use { .. } => ( + crate::diagnostics::get::GET_DIRECT_MEMORY_DECISION_USE, + crate::diagnostics::get::GET_DIRECT_MEMORY_REASON_NONE, + ), + GetDirectMemoryDecision::Fallback(reason) => { + (crate::diagnostics::get::GET_DIRECT_MEMORY_DECISION_FALLBACK, reason.as_str()) + } + }; + rustfs_io_metrics::record_get_object_direct_memory_decision(outcome, object_class.as_str(), reason, size_bucket); +} + fn record_get_object_reader_path_observation( path: &'static str, object_class: GetCodecStreamingObjectClass, @@ -1026,50 +1104,108 @@ fn is_get_small_object_direct_memory_eligible_with_threshold( opts: &ObjectOptions, threshold: usize, ) -> bool { - if threshold == 0 - || range.is_some() - || opts.part_number.is_some() - || opts.version_id.is_some() - || opts.versioned - || opts.version_suspended - || opts.incl_free_versions - || opts.skip_free_version - || opts.data_movement - || opts.raw_data_movement_read - || object_info.delete_marker - || object_info.metadata_only - || object_info.version_only - || object_info.is_encrypted() - || object_info.is_compressed() - || object_info.is_remote() - || object_info.parts.len() != 1 - || fi.parts.len() != 1 - || fi.size <= 0 - { - return false; - } - - let Ok(object_size) = usize::try_from(fi.size) else { - return false; - }; - - object_size <= threshold + matches!( + get_small_object_direct_memory_decision_with_threshold(range, object_info, fi, opts, true, threshold), + GetDirectMemoryDecision::Use { .. } + ) } -fn is_get_small_object_direct_memory_eligible( +fn get_small_object_direct_memory_decision_with_threshold( range: &Option, object_info: &ObjectInfo, fi: &FileInfo, opts: &ObjectOptions, -) -> bool { - is_get_small_object_direct_memory_enabled() - && is_get_small_object_direct_memory_eligible_with_threshold( - range, - object_info, - fi, - opts, - get_small_object_direct_memory_threshold(), - ) + enabled: bool, + threshold: usize, +) -> GetDirectMemoryDecision { + if !enabled { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Disabled); + } + if threshold == 0 { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::ThresholdZero); + } + if range.is_some() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Range); + } + if opts.part_number.is_some() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::PartNumber); + } + if opts.version_id.is_some() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::VersionId); + } + if opts.versioned { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Versioned); + } + if opts.version_suspended { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::VersionSuspended); + } + if opts.incl_free_versions { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InclFreeVersions); + } + if opts.skip_free_version { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SkipFreeVersion); + } + if opts.data_movement { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::DataMovement); + } + if opts.raw_data_movement_read { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::RawDataMovementRead); + } + if object_info.delete_marker { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::DeleteMarker); + } + if object_info.metadata_only { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::MetadataOnly); + } + if object_info.version_only { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::VersionOnly); + } + if object_info.is_encrypted() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Encrypted); + } + if object_info.is_compressed() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Compressed); + } + if object_info.is_remote() { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Remote); + } + if object_info.parts.len() != 1 { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::ObjectInfoMultipart); + } + if fi.parts.len() != 1 { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::FileInfoMultipart); + } + + let Ok(object_size) = usize::try_from(fi.size) else { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize); + }; + if object_size == 0 { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::InvalidSize); + } + if object_info.size != fi.size { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch); + } + if object_size > threshold { + return GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::AboveThreshold); + } + + GetDirectMemoryDecision::Use { object_size } +} + +fn get_small_object_direct_memory_decision( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, +) -> GetDirectMemoryDecision { + get_small_object_direct_memory_decision_with_threshold( + range, + object_info, + fi, + opts, + is_get_small_object_direct_memory_enabled(), + get_small_object_direct_memory_threshold(), + ) } fn should_prefer_codec_streaming_data_blocks_first_reader_setup( @@ -1779,6 +1915,15 @@ fn should_use_single_block_non_inline_fast_path(is_inline_buffer: bool, object_s !is_inline_buffer && object_fits_single_block(object_size, block_size) } +fn should_use_inline_fast_path( + range: &Option, + object_info: &ObjectInfo, + fi: &FileInfo, + opts: &ObjectOptions, +) -> bool { + object_info.is_inline_fast_path_eligible() && fi.data.is_some() && range.is_none() && opts.part_number.is_none() +} + enum SmallWritePath { Inline, SingleBlockNonInline, @@ -2156,8 +2301,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { // Inline data fast path: skip duplex pipe for small inline objects. // Uses the shared predicate from ObjectInfo; additionally checks that - // inline data is actually present and no range request is in flight. - if object_info.is_inline_fast_path_eligible() && fi.data.is_some() && range.is_none() { + // inline data is actually present and neither range nor partNumber is + // in flight. + if should_use_inline_fast_path(&range, &object_info, &fi, opts) { let mut inline_prepare_stage_start = get_stage_timer_if_enabled(stage_metrics_enabled); let data_shards = fi.erasure.data_blocks; @@ -2372,9 +2518,9 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { return Ok(reader); } - if is_get_small_object_direct_memory_eligible(&range, &object_info, &fi, opts) { - let object_size = usize::try_from(object_info.size) - .map_err(|_| to_object_err(Error::other("direct-memory GET object size is invalid"), vec![bucket, object]))?; + let direct_memory_decision = get_small_object_direct_memory_decision(&range, &object_info, &fi, opts); + record_get_direct_memory_decision(object_class, direct_memory_decision, size_bucket); + if let GetDirectMemoryDecision::Use { object_size } = direct_memory_decision { if let Some(body) = Self::try_get_object_direct_data_shards_with_fileinfo( bucket, object, @@ -9594,6 +9740,118 @@ mod tests { )); } + #[test] + fn inline_fast_path_rejects_part_number_requests() { + let (mut object_info, mut fi, opts) = direct_memory_test_metadata(1024); + object_info.inlined = true; + fi.data = Some(Bytes::from(vec![0; 1024])); + + assert!(should_use_inline_fast_path(&None, &object_info, &fi, &opts)); + + let mut part_opts = opts; + part_opts.part_number = Some(1); + assert!(!should_use_inline_fast_path(&None, &object_info, &fi, &part_opts)); + } + + #[test] + fn small_object_direct_memory_decision_reports_bounded_reasons() { + let (object_info, fi, opts) = direct_memory_test_metadata(1024); + + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, false, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Disabled) + ); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 0), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::ThresholdZero) + ); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold( + &Some(HTTPRangeSpec { + start: 0, + end: 10, + is_suffix_length: false, + }), + &object_info, + &fi, + &opts, + true, + 128 * 1024 + ), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Range) + ); + + let mut versioned_opts = opts.clone(); + versioned_opts.versioned = true; + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &versioned_opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Versioned) + ); + + let mut encrypted = object_info.clone(); + Arc::make_mut(&mut encrypted.user_defined).insert("x-amz-server-side-encryption".to_string(), "AES256".to_string()); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &encrypted, &fi, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Encrypted) + ); + + let mut remote = object_info.clone(); + remote.transitioned_object.status = TRANSITION_COMPLETE.to_string(); + remote.transitioned_object.tier = "remote-tier".to_string(); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &remote, &fi, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::Remote) + ); + + let mut multipart = object_info.clone(); + multipart.parts = Arc::new(vec![ObjectPartInfo::default(), ObjectPartInfo::default()]); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &multipart, &fi, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::ObjectInfoMultipart) + ); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 512), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::AboveThreshold) + ); + + let mut size_mismatch = object_info.clone(); + size_mismatch.size += 1; + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &size_mismatch, &fi, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Fallback(GetDirectMemoryFallbackReason::SizeMismatch) + ); + assert_eq!( + get_small_object_direct_memory_decision_with_threshold(&None, &object_info, &fi, &opts, true, 128 * 1024), + GetDirectMemoryDecision::Use { object_size: 1024 } + ); + } + + #[test] + fn direct_memory_fallback_metric_labels_are_stable() { + assert_eq!(GetDirectMemoryFallbackReason::Disabled.as_str(), "disabled"); + assert_eq!(GetDirectMemoryFallbackReason::ThresholdZero.as_str(), "threshold_zero"); + assert_eq!(GetDirectMemoryFallbackReason::Range.as_str(), "range"); + assert_eq!(GetDirectMemoryFallbackReason::PartNumber.as_str(), "part_number"); + assert_eq!(GetDirectMemoryFallbackReason::VersionId.as_str(), "version_id"); + assert_eq!(GetDirectMemoryFallbackReason::Versioned.as_str(), "versioned"); + assert_eq!(GetDirectMemoryFallbackReason::VersionSuspended.as_str(), "version_suspended"); + assert_eq!(GetDirectMemoryFallbackReason::InclFreeVersions.as_str(), "incl_free_versions"); + assert_eq!(GetDirectMemoryFallbackReason::SkipFreeVersion.as_str(), "skip_free_version"); + assert_eq!(GetDirectMemoryFallbackReason::DataMovement.as_str(), "data_movement"); + assert_eq!(GetDirectMemoryFallbackReason::RawDataMovementRead.as_str(), "raw_data_movement_read"); + assert_eq!(GetDirectMemoryFallbackReason::DeleteMarker.as_str(), "delete_marker"); + assert_eq!(GetDirectMemoryFallbackReason::MetadataOnly.as_str(), "metadata_only"); + assert_eq!(GetDirectMemoryFallbackReason::VersionOnly.as_str(), "version_only"); + assert_eq!(GetDirectMemoryFallbackReason::Encrypted.as_str(), "encrypted"); + assert_eq!(GetDirectMemoryFallbackReason::Compressed.as_str(), "compressed"); + assert_eq!(GetDirectMemoryFallbackReason::Remote.as_str(), "remote"); + assert_eq!(GetDirectMemoryFallbackReason::ObjectInfoMultipart.as_str(), "object_info_multipart"); + assert_eq!(GetDirectMemoryFallbackReason::FileInfoMultipart.as_str(), "file_info_multipart"); + assert_eq!(GetDirectMemoryFallbackReason::InvalidSize.as_str(), "invalid_size"); + assert_eq!(GetDirectMemoryFallbackReason::SizeMismatch.as_str(), "size_mismatch"); + assert_eq!(GetDirectMemoryFallbackReason::AboveThreshold.as_str(), "above_threshold"); + } + #[test] fn small_object_direct_memory_eligibility_respects_threshold_and_shape() { let (object_info, fi, opts) = direct_memory_test_metadata(128 * 1024); diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index ae05c2430..a0fb5a729 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -371,6 +371,9 @@ impl MetadataQuorumAccumulator { /// Check if a versioned request can early-stop because the requested /// version_id has reached quorum across disks. fn version_early_stop_decision(&self) -> Option { + if !self.allow_early_stop { + return None; + } if self.requested_version_id.is_empty() { return None; } @@ -4388,6 +4391,18 @@ mod tests { MetadataQuorumAccumulator::new(4, 2, true).with_requested_version_id(requested_version_id) } + #[test] + fn version_early_stop_respects_disabled_accumulator() { + let vid = Uuid::new_v4(); + let mut accumulator = MetadataQuorumAccumulator::new(4, 2, false).with_requested_version_id(&vid.to_string()); + + accumulator.observe_file_info(&version_early_stop_candidate("object", 1, vid)); + accumulator.observe_file_info(&version_early_stop_candidate("object", 2, vid)); + + assert!(accumulator.version_early_stop_decision().is_none()); + assert_eq!(accumulator.matching_version_votes, 2); + } + #[test] fn version_early_stop_hits_quorum_with_matching_versions() { let vid = Uuid::new_v4(); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index b7c2d51ed..e49d85fe7 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -724,6 +724,34 @@ pub fn record_get_object_direct_memory_subpath(subpath: &'static str, object_cla .increment(1); } +/// Record the direct-memory GetObject path decision and bounded fallback reason. +#[inline(always)] +pub fn record_get_object_direct_memory_decision( + outcome: &'static str, + object_class: &'static str, + reason: &'static str, + size_bucket: &'static str, +) { + if !get_stage_metrics_enabled() { + return; + } + counter!( + "rustfs_io_get_object_direct_memory_decision_total", + "outcome" => outcome, + "object_class" => object_class, + "reason" => reason + ) + .increment(1); + counter!( + "rustfs_io_get_object_direct_memory_decision_by_size_total", + "outcome" => outcome, + "object_class" => object_class, + "reason" => reason, + "size_bucket" => size_bucket + ) + .increment(1); +} + /// Record why the codec streaming reader was not selected. #[inline(always)] pub fn record_get_object_codec_streaming_fallback(reason: &'static str) { diff --git a/docs/benchmark/issue824-get-pressure-results-2026-07-05.md b/docs/benchmark/issue824-get-pressure-results-2026-07-05.md new file mode 100644 index 000000000..6e6d3397b --- /dev/null +++ b/docs/benchmark/issue824-get-pressure-results-2026-07-05.md @@ -0,0 +1,86 @@ +# Issue 824 GET Pressure Validation Results + +Date: 2026-07-05 + +## Scope + +This run validates the current branch `houseme/get-object-performance-hardening` +against the closest local historical GET metrics gate baseline. + +Historical baseline: + +- Path: `docs/benchmark/rustfs-target-bench/get-metrics-gate-baseline/warp/median_summary.csv` +- Git head: `92c50156a3cbbdfa454be1d14062098708851f51` +- Parameters: `10MiB`, warp, concurrency `32`, duration `10s`, rounds `3` +- Median throughput: `5734882344.960000` B/s +- Median req/s: `546.920000` +- Median latency: `48.700000` ms + +Current run: + +- Path: `target/bench/issue824-get-metrics-gate-current-20260705` +- Git head recorded by script: `55ad8df1c2f574178669052093aa185fae8185ca` +- Branch/worktree contained staged and unstaged issue-824 changes. +- Parameters: `10MiB`, warp, concurrency `32`, duration `10s`, rounds `3` +- RustFS binary: `target/release/rustfs` +- Observability export: disabled +- Scanner: disabled +- Command: + +```bash +scripts/run_get_metrics_gate_smoke.sh \ + --skip-build \ + --rustfs-bin target/release/rustfs \ + --size 10MiB \ + --concurrency 32 \ + --duration 10s \ + --rounds 3 \ + --round-cooldown-secs 1 \ + --baseline-csv docs/benchmark/rustfs-target-bench/get-metrics-gate-baseline/warp/median_summary.csv \ + --out-dir target/bench/issue824-get-metrics-gate-current-20260705 \ + --data-root /private/tmp/issue824-get-metrics-gate-current-20260705 +``` + +## Raw Rounds + +| round | throughput | req/s | avg latency | p90 | p99 | +| --- | ---: | ---: | ---: | ---: | ---: | +| 1 | 6505.75 MiB/s | 650.57 | 47.9 ms | 73.0 ms | 103.9 ms | +| 2 | 5631.35 MiB/s | 563.13 | 52.1 ms | 80.1 ms | 106.3 ms | +| 3 | 5830.30 MiB/s | 583.03 | 52.9 ms | 72.2 ms | 92.7 ms | + +Corrected median from raw round data: + +- Median throughput: `6113512652.800000` B/s (`5830.30 MiB/s`) +- Median req/s: `583.030000` +- Median latency: `52.100000` ms +- Median p90: `73.000000` ms +- Median p99: `103.900000` ms + +## Baseline Comparison + +| metric | baseline | current | delta | +| --- | ---: | ---: | ---: | +| throughput B/s | 5734882344.960000 | 6113512652.800000 | +6.60% | +| req/s | 546.920000 | 583.030000 | +6.60% | +| avg latency ms | 48.700000 | 52.100000 | +6.98% | + +Interpretation: + +- No throughput regression was observed in this local release-binary smoke. +- Latency moved higher by 6.98% against the selected baseline. This is not a + blocking regression by itself because host state, cooldown, and branch state + differ, but it should be watched in the PR benchmark matrix. +- The run validates that the current changes do not obviously collapse the + default 10MiB/C32 GET hot path. + +## Artifact Caveat + +The script completed with exit code `0`, but its generated +`warp/median_summary.csv` and `warp/baseline_compare.csv` contain `N/A` values. +The terminal summary and `warp/round_results.csv` preserved the actual round +values. The corrected medians above are computed from those raw round logs. + +Follow-up: fix `scripts/run_object_batch_bench_enhanced.sh` CSV parsing/report +generation so the persisted median and baseline comparison match the terminal +summary. diff --git a/docs/benchmark/issue829-default-get-hot-path-runbook-2026-07-05.md b/docs/benchmark/issue829-default-get-hot-path-runbook-2026-07-05.md new file mode 100644 index 000000000..4621a9ac8 --- /dev/null +++ b/docs/benchmark/issue829-default-get-hot-path-runbook-2026-07-05.md @@ -0,0 +1,283 @@ +# Issue 829 Default GET Hot Path Benchmark and Attribution Runbook + +Date: 2026-07-05 + +## Scope + +This runbook defines the evidence package for proving the current default GET +hot path. It intentionally separates throughput evidence from path attribution: + +- Metrics-off runs are throughput evidence. +- Metrics-on diagnostic runs are path attribution evidence. +- Codec and direct-memory experiments are A/B evidence, not default-path proof + unless their enabling environment is recorded. + +The current review baseline is: + +- Ordinary non-inline GETs are expected to use ECStore metadata/cache lookup and + the legacy duplex reader path. +- Metadata early-stop is not expected for normal `read_data` GETs. +- Direct-memory, codec streaming, and metadata early-stop are opt-in or gated. +- Diagnostic metrics add overhead and must not be used as pure throughput proof. +- Metadata cache is topology-sensitive: distributed erasure currently bypasses + the metadata GET cache, so distributed baselines should assume full metadata + fanout unless a later change records otherwise. +- Rollout percentage is not the primary safety gate for codec streaming or + metadata early-stop in the current defaults; record the base enable flag and + compatibility switches before interpreting any percentage value. +- Codec-streaming cost accounting must include both shard-to-output-buffer and + output-buffer-to-client copies, plus the `+ Sync` reader wrapper lock cost. + +## Artifact Contract + +Every run directory must keep enough information to reproduce or reject the +result: + +- `git_head` or `git rev-parse HEAD` +- branch name and dirty-tree status +- full script command and arguments +- RustFS binary path and build mode +- object size, object class, bucket, concurrency, duration, rounds +- metric mode: metrics-off throughput or metrics-on attribution +- environment affecting GET path selection +- `manifest.env` and/or `environment.txt` +- `warp/median_summary.csv` and `warp/round_results.csv` for throughput runs +- `service_metrics_summary.csv`, `service_metrics_stage_distribution.csv`, or + raw before/after Prometheus snapshots for attribution runs + +Do not compare a metrics-on run against a metrics-off run as a throughput win or +loss. The only valid cross-mode claim is attribution. + +## Baseline Dimensions + +Object sizes: + +- `1KiB` +- `4KiB` +- `10KiB` +- `64KiB` +- `100KiB` +- `128KiB` +- `1MiB` +- `10MiB` +- `128MiB` + +Object classes: + +- default inline candidate +- forced non-inline candidate +- plain single-part +- multipart +- range request +- versioned bucket +- checksum-mode request +- CORS request + +Run the small default-path proof first. Expand into large, range, multipart, and +versioned dimensions only after the small matrix has stable artifacts. + +## Step 0: Script Syntax Gate + +```bash +for s in \ + scripts/run_get_codec_streaming_smoke.sh \ + scripts/run_get_metrics_gate_smoke.sh \ + scripts/run_object_batch_bench_enhanced.sh \ + scripts/run_object_data_cache_bench.sh +do + bash -n "$s" +done +``` + +## Step 1: Metrics-Off Throughput Smoke + +Use this run only for throughput and latency. Observability export and detailed +GET stage attribution are disabled by the script. + +```bash +scripts/run_get_metrics_gate_smoke.sh \ + --skip-build \ + --size 1KiB \ + --concurrency 32 \ + --duration 10s \ + --rounds 3 \ + --out-dir target/bench/get-default-metrics-off-1kib-c32 +``` + +Required checks: + +- `target/bench/get-default-metrics-off-1kib-c32/manifest.env` records + observability export as disabled. +- `warp/median_summary.csv` and `warp/round_results.csv` exist. +- No statement about reader path is made from this run alone. + +## Step 2: Metrics-On Default Path Attribution + +Use this run to prove the default reader path. It may be slower than the +metrics-off smoke and that is expected. + +```bash +scripts/run_get_codec_streaming_smoke.sh \ + --mode legacy \ + --sizes 1KiB,4KiB,10MiB \ + --concurrency 32 \ + --duration 20s \ + --rounds 3 \ + --diagnostic-metrics \ + --diagnostic-metrics-filter-regex 'rustfs[._]get_object[._](reader_path|stage_duration|reader_setup|disk_permit|request|reader)' \ + --out-dir target/bench/get-default-attribution-legacy +``` + +Required checks: + +- `manifest.env` records `mode=legacy` and diagnostic metrics enabled. +- `service_metrics_summary.csv` exists. +- The default path proof reports the delta for + `rustfs_io_get_object_reader_path_total{path="legacy_duplex"}`. +- If `legacy_duplex` is not positive, report the observed path instead of + forcing the expected conclusion. + +## Step 3: Small-Object Matrix + +Run the small sizes with metrics-off first: + +```bash +for size in 1KiB 4KiB 10KiB 64KiB 100KiB 128KiB 1MiB; do + scripts/run_get_metrics_gate_smoke.sh \ + --skip-build \ + --size "$size" \ + --concurrency 32 \ + --duration 10s \ + --rounds 3 \ + --out-dir "target/bench/get-default-metrics-off-${size}-c32" +done +``` + +Then run attribution only for representative sizes: + +```bash +scripts/run_get_codec_streaming_smoke.sh \ + --mode legacy \ + --sizes 1KiB,10KiB,128KiB,1MiB \ + --concurrency 32 \ + --duration 20s \ + --rounds 3 \ + --diagnostic-metrics \ + --diagnostic-metrics-filter-regex 'rustfs[._]get_object[._](reader_path|stage_duration|reader_setup|disk_permit|request|reader)' \ + --out-dir target/bench/get-default-small-attribution +``` + +## Step 4: Large Sequential and Range Matrix + +Use the dedicated GT1G helper for very large sequential/ranged GETs when the +target machine has enough local disk and time: + +```bash +scripts/run_gt1g_get_http_matrix.sh \ + --skip-build \ + --out-dir target/bench/get-default-large-http-matrix +``` + +For a shorter release-gate matrix: + +```bash +scripts/run_get_codec_streaming_smoke.sh \ + --mode legacy \ + --sizes 10MiB,128MiB \ + --concurrency 16 \ + --duration 20s \ + --rounds 3 \ + --out-dir target/bench/get-default-large-metrics-off +``` + +Add `--diagnostic-metrics` only in a separate attribution run. + +## Step 5: Codec Streaming A/B Matrix + +Codec streaming is not default-path proof. Use it only after the default legacy +path has been proven. + +```bash +scripts/run_get_codec_streaming_smoke.sh \ + --mode both \ + --profile-order reverse \ + --codec-engine rustfs \ + --sizes 1MiB,10MiB,128MiB \ + --concurrency 16 \ + --duration 20s \ + --rounds 3 \ + --diagnostic-metrics \ + --out-dir target/bench/get-codec-ab-attribution +``` + +Required checks: + +- Legacy profile has a positive `legacy_duplex` path delta. +- Codec profile has a positive `codec_streaming` path delta only when codec + settings are enabled. +- Compatibility summaries pass before any performance conclusion is accepted. + +## Step 6: Admission and Permit Fallback Stress + +Use this to validate workload admission visibility under pressure: + +```bash +RUSTFS_OBJECT_MAX_CONCURRENT_DISK_READS=1 \ +RUSTFS_OBJECT_DISK_PERMIT_WAIT_TIMEOUT=1 \ +scripts/run_get_codec_streaming_smoke.sh \ + --mode legacy \ + --sizes 10MiB \ + --concurrency 64 \ + --duration 20s \ + --rounds 3 \ + --diagnostic-metrics \ + --diagnostic-metrics-filter-regex 'rustfs[._]get_object[._]disk_permit[._]bypass|rustfs_io_disk_permit|rustfs_io_queue_|rustfs_io_get_object_' \ + --out-dir target/bench/get-permit-admission +``` + +Required checks: + +- `rustfs.get_object.disk_permit.bypass.total` or its exported equivalent is + captured when fallback happens. +- Queue/admission metrics are present in the attribution snapshot. +- Slow-client behavior does not deadlock the benchmark. + +## Reporting Template + +Use this structure in the final issue or PR comment: + +```text +Commit: +Branch: +Dirty tree: +Host: +RustFS binary: +Metric mode: +Command: +Out dir: + +Throughput evidence: +- median req/s: +- median latency: +- p90: +- p99: + +Attribution evidence: +- reader_path legacy_duplex delta: +- reader_path codec_streaming delta: +- top stage durations: +- disk permit fallback delta: + +Conclusion: +- Default path: +- Compatibility: +- Follow-up: +``` + +## Acceptance Checklist + +- Metrics-off throughput and metrics-on attribution are separate directories. +- Every artifact directory records commit, env, object layout, and arguments. +- Default path hit rate is explicitly reported from diagnostic metrics. +- Any default-path claim cites the metric delta used to prove it. +- Codec/direct-memory/metadata-early-stop claims include their enabling env. diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index c5bc53de3..11df0aa54 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -343,8 +343,6 @@ struct GetObjectReadSetup { sse_customer_key_md5: Option, ssekms_key_id: Option, encryption_applied: bool, - /// `true` when the object was read via the inline data fast path (no disk I/O). - is_inline_fast_path: bool, } struct GetObjectPreparedRead { @@ -505,13 +503,48 @@ pin_project! { } } -pin_project! { - struct MemoryTrackedBytesStream { - bytes: Bytes, - emitted: bool, - started: std::time::Instant, - source: &'static str, - _guard: Option, +struct MemoryTrackedBytesStream { + bytes: Bytes, + emitted: bool, + completed: bool, + expected: usize, + started: std::time::Instant, + source: &'static str, + _guard: Option, + lifecycle: GetObjectBodyLifecycle, +} + +#[derive(Default)] +struct GetObjectBodyLifecycle { + request_guard: Option, +} + +impl GetObjectBodyLifecycle { + fn tracked(request_guard: GetObjectGuard) -> Self { + Self { + request_guard: Some(request_guard), + } + } + + #[cfg(test)] + fn disabled() -> Self { + Self { request_guard: None } + } + + fn is_finished(&self) -> bool { + self.request_guard.is_none() + } + + fn finish_ok(&mut self) { + if let Some(mut request_guard) = self.request_guard.take() { + request_guard.finish_ok(); + } + } + + fn finish_err(&mut self) { + if let Some(mut request_guard) = self.request_guard.take() { + request_guard.finish_err(); + } } } @@ -566,15 +599,33 @@ pin_project! { } impl MemoryTrackedBytesStream { - fn new(bytes: Bytes, source: &'static str, guard: Option) -> Self { + fn new( + bytes: Bytes, + expected: usize, + source: &'static str, + guard: Option, + lifecycle: GetObjectBodyLifecycle, + ) -> Self { Self { bytes, emitted: false, + completed: expected == 0, + expected, started: std::time::Instant::now(), source, _guard: guard, + lifecycle, } } + + fn finish_ok(&mut self) { + self.completed = true; + self.lifecycle.finish_ok(); + } + + fn finish_err(&mut self) { + self.lifecycle.finish_err(); + } } impl GetObjectReaderStream @@ -600,9 +651,9 @@ impl futures::Stream for MemoryTrackedBytesStream { type Item = std::io::Result; fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - let this = self.project(); + let this = self.get_mut(); let poll_start = is_get_output_handoff_attribution_enabled().then(std::time::Instant::now); - if *this.emitted { + if this.emitted { if let Some(poll_start) = poll_start { rustfs_io_metrics::record_get_object_memory_body_stream_poll( this.source, @@ -615,10 +666,13 @@ impl futures::Stream for MemoryTrackedBytesStream { } let first_byte_elapsed = (!this.bytes.is_empty()).then(|| this.started.elapsed()); - *this.emitted = true; + this.emitted = true; if let Some(elapsed) = first_byte_elapsed { rustfs_io_metrics::record_get_object_first_byte_latency(GET_OBJECT_STAGE_PATH_S3_HANDLER, elapsed.as_secs_f64()); } + if this.bytes.len() >= this.expected { + this.finish_ok(); + } if let Some(poll_start) = poll_start { rustfs_io_metrics::record_get_object_memory_body_stream_poll( this.source, @@ -631,6 +685,20 @@ impl futures::Stream for MemoryTrackedBytesStream { } } +impl Drop for MemoryTrackedBytesStream { + fn drop(&mut self) { + if self.lifecycle.is_finished() { + return; + } + + if self.completed { + self.finish_ok(); + } else { + self.finish_err(); + } + } +} + impl futures::Stream for GetObjectReaderStream where R: AsyncRead, @@ -730,11 +798,12 @@ struct GetObjectStreamingReader { started: std::time::Instant, first_byte_reported: bool, completed: bool, + lifecycle: GetObjectBodyLifecycle, _foreground_read_guard: rustfs_scanner::ForegroundReadGuard, } impl GetObjectStreamingReader { - fn new(inner: R, bucket: &str, key: &str, expected: usize, timeout: Duration) -> Self { + fn new(inner: R, bucket: &str, key: &str, expected: usize, timeout: Duration, lifecycle: GetObjectBodyLifecycle) -> Self { Self { inner, bucket: bucket.to_string(), @@ -746,6 +815,7 @@ impl GetObjectStreamingReader { started: std::time::Instant::now(), first_byte_reported: false, completed: expected == 0, + lifecycle, _foreground_read_guard: rustfs_scanner::ForegroundReadGuard::new(), } } @@ -753,6 +823,15 @@ impl GetObjectStreamingReader { fn elapsed(&self) -> Duration { self.started.elapsed() } + + fn finish_ok(&mut self) { + self.completed = true; + self.lifecycle.finish_ok(); + } + + fn finish_err(&mut self) { + self.lifecycle.finish_err(); + } } impl AsyncRead for GetObjectStreamingReader { @@ -789,6 +868,7 @@ impl AsyncRead for GetObjectStreamingReader { } if self.emitted >= self.expected { self.completed = true; + self.finish_ok(); } } else if self.emitted < self.expected { warn!( @@ -803,14 +883,17 @@ impl AsyncRead for GetObjectStreamingReader { state = "short_eof", "GetObject streaming body ended before expected length" ); + self.finish_err(); } else { self.completed = true; + self.finish_ok(); } Poll::Ready(Ok(())) } Poll::Ready(Err(err)) => { self.timer = None; + self.finish_err(); warn!( event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, @@ -852,6 +935,7 @@ impl AsyncRead for GetObjectStreamingReader { state = "stall_timeout", "GetObject streaming body stalled" ); + self.finish_err(); return Poll::Ready(Err(std::io::Error::new( std::io::ErrorKind::TimedOut, "get object streaming body stall timeout", @@ -866,10 +950,16 @@ impl AsyncRead for GetObjectStreamingReader { impl Drop for GetObjectStreamingReader { fn drop(&mut self) { - if self.expected == 0 || self.completed || self.emitted >= self.expected { + if self.lifecycle.is_finished() { return; } + if self.expected == 0 || self.completed || self.emitted >= self.expected { + self.finish_ok(); + return; + } + + self.finish_err(); warn!( event = EVENT_GET_OBJECT_STREAM_BODY, component = LOG_COMPONENT_APP, @@ -2095,16 +2185,19 @@ impl DefaultObjectUsecase { fn build_memory_bytes_blob( bytes: Bytes, response_content_length: i64, - _optimal_buffer_size: usize, source: &'static str, - ) -> Option { + lifecycle: GetObjectBodyLifecycle, + ) -> StreamingBlob { let get_stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); let memory_blob_start = get_stage_metrics_enabled.then(std::time::Instant::now); let handoff_start = get_stage_metrics_enabled.then(std::time::Instant::now); let bytes_len = bytes.len(); let guard = rustfs_io_metrics::track_get_object_buffered_bytes(bytes_len); let remaining = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX); - let blob = StreamingBlob::wrap(bytes_stream(MemoryTrackedBytesStream::new(bytes, source, guard), remaining)); + let blob = StreamingBlob::wrap(bytes_stream( + MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle), + remaining, + )); if let Some(handoff_start) = handoff_start { rustfs_io_metrics::record_get_object_response_handoff( "single_chunk", @@ -2115,16 +2208,16 @@ impl DefaultObjectUsecase { ); } record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_MEMORY_BLOB, memory_blob_start); - Some(blob) + blob } fn build_memory_blob( buf: Vec, response_content_length: i64, - optimal_buffer_size: usize, source: &'static str, - ) -> Option { - Self::build_memory_bytes_blob(Bytes::from(buf), response_content_length, optimal_buffer_size, source) + lifecycle: GetObjectBodyLifecycle, + ) -> StreamingBlob { + Self::build_memory_bytes_blob(Bytes::from(buf), response_content_length, source, lifecycle) } fn select_stream_buffer_strategy( @@ -2151,7 +2244,8 @@ impl DefaultObjectUsecase { stream_strategy: GetObjectStreamStrategy, bucket: &str, key: &str, - ) -> Option + lifecycle: GetObjectBodyLifecycle, + ) -> StreamingBlob where R: AsyncRead + Send + Sync + Unpin + 'static, { @@ -2170,7 +2264,7 @@ impl DefaultObjectUsecase { ); } let handoff_start = get_stage_metrics_enabled.then(std::time::Instant::now); - let reader = GetObjectStreamingReader::new(reader, bucket, key, expected, get_object_disk_read_timeout()); + let reader = GetObjectStreamingReader::new(reader, bucket, key, expected, get_object_disk_read_timeout(), lifecycle); let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source); let blob = StreamingBlob::new(stream); if let Some(handoff_start) = handoff_start { @@ -2183,7 +2277,7 @@ impl DefaultObjectUsecase { ); } record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_STREAMING_BLOB, streaming_blob_start); - Some(blob) + blob } fn init_get_object_bootstrap(bucket: &str, key: &str, request_id: &str) -> S3Result { @@ -2200,8 +2294,6 @@ impl DefaultObjectUsecase { Self::ensure_get_object_not_timed_out(&wrapper, &timeout_config, bucket, key, GetObjectTimeoutStage::BeforeProcessing)?; - rustfs_io_metrics::record_get_object_request_start(concurrent_requests); - debug!( "GetObject request started with {} concurrent requests, timeout={:?}", concurrent_requests, timeout_config.get_object_timeout @@ -2217,6 +2309,16 @@ impl DefaultObjectUsecase { }) } + fn validate_get_object_part_number(part_number: Option, info: &ObjectInfo) -> S3Result<()> { + if let Some(part_number) = part_number + && part_number > 1 + && !info.parts.iter().any(|part| part.number == part_number) + { + return Err(s3_error!(InvalidPart)); + } + Ok(()) + } + /// How long a GET waits for a disk read permit before degrading to a /// permit-less read. Cached: consulted per GET. Zero disables the bound. fn disk_permit_wait_timeout() -> Duration { @@ -2313,13 +2415,7 @@ impl DefaultObjectUsecase { validate_object_key(&key, "GET")?; - let part_number = part_number.map(|v| v as usize); - - if let Some(part_num) = part_number - && part_num == 0 - { - return Err(s3_error!(InvalidArgument, "Invalid part number: part number must be greater than 0")); - } + let part_number = parse_part_number_i32_to_usize(part_number, "GET")?; let rs = range.map(|v| match v { Range::Int { first, last } => HTTPRangeSpec { @@ -2376,8 +2472,14 @@ impl DefaultObjectUsecase { ); } - // SF05: Read object metadata/data BEFORE acquiring disk I/O semaphore. - // ECStore's get_object_reader acquires its own RwLock — safe without the semaphore. + // Acquire the GET disk permit before ECStore reader setup. Reader setup + // may perform metadata fanout, materialize direct-memory bodies, or + // start legacy duplex background reads; all of that work belongs inside + // the admission boundary. Fully materialized bodies release the permit + // when the unused wrapped reader is dropped during body construction, + // while streaming bodies keep it until EOF or client drop. + let io_planning = Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await?; + let read_start = std::time::Instant::now(); let read_stage_start = rustfs_io_metrics::get_stage_metrics_enabled().then_some(read_start); let read_setup = Self::prepare_get_object_read( @@ -2395,18 +2497,6 @@ impl DefaultObjectUsecase { ) .await?; - // SF05: Skip disk I/O semaphore for inline fast path — data is already in memory. - let io_planning = if read_setup.is_inline_fast_path { - GetObjectIoPlanning { - disk_permit: None, - permit_wait_duration: Duration::ZERO, - queue_status: concurrency::IoQueueStatus::default(), - queue_utilization: 0.0, - } - } else { - Self::acquire_get_object_io_planning(manager, wrapper, timeout_config, bucket, key).await? - }; - Ok(GetObjectPreparedRead { io_planning, read_setup }) } @@ -2450,6 +2540,7 @@ impl DefaultObjectUsecase { } 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() { @@ -2544,10 +2635,6 @@ impl DefaultObjectUsecase { None => (None, None, None, None, false, wrap_reader(stream), buffered_body), }; - // Detect inline fast path: data is in memory, no disk I/O semaphore needed. - // Uses the shared predicate from ObjectInfo; additionally checks no range request. - let is_inline_fast_path = info.is_inline_fast_path_eligible() && rs.is_none(); - Ok(GetObjectReadSetup { info, final_stream, @@ -2562,7 +2649,6 @@ impl DefaultObjectUsecase { sse_customer_key_md5, ssekms_key_id, encryption_applied, - is_inline_fast_path, }) } #[allow(clippy::too_many_arguments)] @@ -2722,7 +2808,8 @@ impl DefaultObjectUsecase { buffered_body: Option, bucket: &str, key: &str, - ) -> S3Result> + mut lifecycle: GetObjectBodyLifecycle, + ) -> S3Result where R: AsyncRead + Send + Sync + Unpin + 'static, { @@ -2736,6 +2823,7 @@ impl DefaultObjectUsecase { let read_result = tokio::io::AsyncReadExt::read_to_end(&mut final_stream, &mut buf).await; record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_ENCRYPTED_BUFFER_READ, buffer_read_start); if let Err(e) = read_result { + lifecycle.finish_err(); error!(error = %e, "GetObject decrypted object buffering failed"); return Err(ApiError::from(StorageError::other(format!("Failed to read decrypted object: {e}"))).into()); } @@ -2751,8 +2839,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_blob( buf, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_ENCRYPTED_BUFFER, + lifecycle, )); } @@ -2768,6 +2856,7 @@ impl DefaultObjectUsecase { stream_strategy, bucket, key, + lifecycle, )); } @@ -2783,8 +2872,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_bytes_blob( buffered_body, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + lifecycle, )); } @@ -2809,8 +2898,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_blob( buf, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_SEEK_BUFFER, + lifecycle, )); } Err(e) => { @@ -2830,6 +2919,7 @@ impl DefaultObjectUsecase { stream_strategy, bucket, key, + lifecycle, )) } @@ -2848,7 +2938,8 @@ impl DefaultObjectUsecase { buffered_body: Option, bucket: &str, key: &str, - ) -> S3Result> + mut lifecycle: GetObjectBodyLifecycle, + ) -> S3Result where R: AsyncRead + Send + Sync + Unpin + 'static, { @@ -2868,8 +2959,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_bytes_blob( bytes, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE, + lifecycle, )); } GetObjectBodyCacheLookup::Disabled | GetObjectBodyCacheLookup::Skip | GetObjectBodyCacheLookup::Miss => {} @@ -2881,8 +2972,8 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_bytes_blob( buffered_body, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + lifecycle, )); } @@ -2915,6 +3006,7 @@ impl DefaultObjectUsecase { None, bucket, key, + lifecycle, ) .await; }; @@ -2940,11 +3032,12 @@ impl DefaultObjectUsecase { return Ok(Self::build_memory_bytes_blob( bytes, response_content_length, - optimal_buffer_size, GET_MEMORY_BODY_SOURCE_OBJECT_DATA_CACHE_MATERIALIZED, + lifecycle, )); } Err(e) => { + lifecycle.finish_err(); error!(error = %e, "GetObject materialize-fill buffering failed"); // The stream is partially consumed; falling back to the // streaming path would send a body missing its prefix, so @@ -2970,6 +3063,7 @@ impl DefaultObjectUsecase { None, bucket, key, + lifecycle, ) .await } @@ -3687,6 +3781,7 @@ impl DefaultObjectUsecase { concurrent_requests: usize, part_number: Option, versioned: bool, + lifecycle: GetObjectBodyLifecycle, ) -> S3Result { let strategy_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); let strategy = self.finalize_get_object_strategy( @@ -3724,6 +3819,7 @@ impl DefaultObjectUsecase { buffered_body, bucket, key, + lifecycle, ) .await?; record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_BUILD, body_build_start); @@ -3762,7 +3858,7 @@ impl DefaultObjectUsecase { record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_METADATA_FILTER, metadata_filter_start); let output = GetObjectOutput { - body, + body: Some(body), content_length: Some(response_content_length), last_modified, content_type, @@ -3825,13 +3921,19 @@ impl DefaultObjectUsecase { let wrapper = bootstrap.wrapper; let request_start = bootstrap.request_start; let concurrent_requests = bootstrap.concurrent_requests; - let mut request_guard = bootstrap.request_guard; + let mut lifecycle = GetObjectBodyLifecycle::tracked(bootstrap.request_guard); let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event(); // mc get 3 let request_context_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now); - let request_context = Self::prepare_get_object_request_context(&req).await?; + let request_context = match Self::prepare_get_object_request_context(&req).await { + Ok(request_context) => request_context, + Err(err) => { + lifecycle.finish_err(); + return Err(err); + } + }; if let Some(request_context_start) = request_context_start { rustfs_io_metrics::record_get_object_stage_duration( "s3_handler", @@ -3850,7 +3952,7 @@ impl DefaultObjectUsecase { let manager = get_concurrency_manager(); - let prepared_read = Self::prepare_get_object_read_execution( + let prepared_read = match Self::prepare_get_object_read_execution( &req, manager, &wrapper, @@ -3861,7 +3963,14 @@ impl DefaultObjectUsecase { &opts, part_number, ) - .await?; + .await + { + Ok(prepared_read) => prepared_read, + Err(err) => { + lifecycle.finish_err(); + return Err(err); + } + }; let GetObjectPreparedRead { io_planning, read_setup } = prepared_read; let GetObjectIoPlanning { disk_permit, @@ -3884,7 +3993,6 @@ impl DefaultObjectUsecase { sse_customer_key_md5, ssekms_key_id, encryption_applied, - is_inline_fast_path: _, } = read_setup; let final_stream = if let Some(disk_permit) = disk_permit { wrap_reader(DiskReadPermitReader::new(final_stream, disk_permit)) @@ -3923,8 +4031,13 @@ impl DefaultObjectUsecase { concurrent_requests, part_number, opts.versioned, + lifecycle, ) - .await?; + .await; + let output_context = match output_context { + Ok(output_context) => output_context, + Err(err) => return Err(err), + }; if let Some(output_build_start) = output_build_start { rustfs_io_metrics::record_get_object_stage_duration( "s3_handler", @@ -3948,22 +4061,8 @@ impl DefaultObjectUsecase { optimal_buffer_size, ); - let result = Self::finalize_get_object_response( - helper, - &bucket, - &req.method, - &req.headers, - event_info, - version_id_for_event, - output, - ) - .await; - if result.is_ok() { - request_guard.finish_ok(); - } else { - request_guard.finish_err(); - } - result + Self::finalize_get_object_response(helper, &bucket, &req.method, &req.headers, event_info, version_id_for_event, output) + .await } pub async fn execute_get_object_attributes( @@ -6624,7 +6723,14 @@ mod tests { #[tokio::test] async fn get_object_streaming_reader_times_out_when_body_stalls() { - let reader = GetObjectStreamingReader::new(PendingReader, "test-bucket", "stalled-object", 1, Duration::from_millis(1)); + let reader = GetObjectStreamingReader::new( + PendingReader, + "test-bucket", + "stalled-object", + 1, + Duration::from_millis(1), + GetObjectBodyLifecycle::disabled(), + ); let mut stream = ReaderStream::with_capacity(reader, 1024); let err = stream @@ -6636,6 +6742,97 @@ mod tests { assert_eq!(err.kind(), std::io::ErrorKind::TimedOut); } + #[tokio::test] + #[serial_test::serial] + async fn get_object_streaming_reader_holds_request_guard_until_eof() { + use tokio::io::AsyncReadExt; + + let initial = GetObjectGuard::concurrent_count(); + let guard = GetObjectGuard::new(); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + + let mut reader = GetObjectStreamingReader::new( + std::io::Cursor::new(b"hello".to_vec()), + "test-bucket", + "complete-object", + 5, + Duration::ZERO, + GetObjectBodyLifecycle::tracked(guard), + ); + let mut out = Vec::new(); + + reader + .read_to_end(&mut out) + .await + .expect("complete streaming body should read successfully"); + + assert_eq!(out, b"hello"); + assert_eq!(GetObjectGuard::concurrent_count(), initial); + } + + #[test] + #[serial_test::serial] + fn get_object_streaming_reader_releases_request_guard_when_dropped_incomplete() { + let initial = GetObjectGuard::concurrent_count(); + let guard = GetObjectGuard::new(); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + + let reader = GetObjectStreamingReader::new( + std::io::Cursor::new(b"short".to_vec()), + "test-bucket", + "dropped-object", + 10, + Duration::ZERO, + GetObjectBodyLifecycle::tracked(guard), + ); + drop(reader); + + assert_eq!(GetObjectGuard::concurrent_count(), initial); + } + + #[tokio::test] + #[serial_test::serial] + async fn memory_tracked_bytes_stream_releases_request_guard_after_emit() { + let initial = GetObjectGuard::concurrent_count(); + let guard = GetObjectGuard::new(); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + + let mut stream = MemoryTrackedBytesStream::new( + Bytes::from_static(b"hello"), + 5, + GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + None, + GetObjectBodyLifecycle::tracked(guard), + ); + let chunk = stream + .next() + .await + .expect("memory body should emit one chunk") + .expect("memory body chunk should be readable"); + + assert_eq!(chunk.as_ref(), b"hello"); + assert_eq!(GetObjectGuard::concurrent_count(), initial); + } + + #[test] + #[serial_test::serial] + fn memory_tracked_bytes_stream_releases_request_guard_for_zero_length_without_poll() { + let initial = GetObjectGuard::concurrent_count(); + let guard = GetObjectGuard::new(); + assert_eq!(GetObjectGuard::concurrent_count(), initial + 1); + + let stream = MemoryTrackedBytesStream::new( + Bytes::new(), + 0, + GET_MEMORY_BODY_SOURCE_BUFFERED_BODY, + None, + GetObjectBodyLifecycle::tracked(guard), + ); + drop(stream); + + assert_eq!(GetObjectGuard::concurrent_count(), initial); + } + #[tokio::test] async fn disk_read_permit_reader_holds_permit_until_reader_is_dropped() { let semaphore = Arc::new(tokio::sync::Semaphore::new(1)); @@ -6663,7 +6860,7 @@ mod tests { ..Default::default() }; - let body = DefaultObjectUsecase::build_get_object_body( + let _body = DefaultObjectUsecase::build_get_object_body( reader, &info, 18_i64 * 1024 * 1024 * 1024, @@ -6676,11 +6873,11 @@ mod tests { None, "test-bucket", "large-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("build_get_object_body should succeed for streaming path"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -6699,7 +6896,7 @@ mod tests { ..Default::default() }; - let body = DefaultObjectUsecase::build_get_object_body( + let _body = DefaultObjectUsecase::build_get_object_body( reader, &info, 18_i64 * 1024 * 1024 * 1024, @@ -6712,11 +6909,11 @@ mod tests { None, "test-bucket", "large-encrypted-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("build_get_object_body should succeed for encrypted streaming path"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -6735,7 +6932,7 @@ mod tests { ..Default::default() }; - let body = DefaultObjectUsecase::build_get_object_body( + let _body = DefaultObjectUsecase::build_get_object_body( reader, &info, 4, @@ -6748,11 +6945,11 @@ mod tests { Some(Bytes::from_static(b"test")), "test-bucket", "direct-memory-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("build_get_object_body should consume buffered body"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -6790,7 +6987,7 @@ mod tests { assert_eq!(fill, rustfs_object_data_cache::ObjectDataCacheFillResult::Inserted); - let body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, reader, &info, @@ -6804,11 +7001,11 @@ mod tests { None, "test-bucket", "cached-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("cache hit body handoff should succeed"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -6844,7 +7041,7 @@ mod tests { }); let fill = adapter.cache().fill_body(&plan, Bytes::from_static(b"oops")).await; - let body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, reader, &info, @@ -6858,13 +7055,13 @@ mod tests { None, "test-bucket", "cached-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("size-mismatched direct fill should not create a cache hit"); let lookup_after_mismatch = adapter.lookup_body(&plan).await; assert_eq!(fill, rustfs_object_data_cache::ObjectDataCacheFillResult::SkippedSizeMismatch); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -6899,7 +7096,7 @@ mod tests { }) .expect("fill-enabled cache adapter should initialize"); - let first_body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _first_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, first_reader, &info, @@ -6913,11 +7110,12 @@ mod tests { Some(Bytes::from_static(b"hello")), "test-bucket", "cached-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("buffered-body handoff should succeed"); - let second_body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _second_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, second_reader, &info, @@ -6931,12 +7129,11 @@ mod tests { None, "test-bucket", "cached-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("follow-up cache hit should succeed"); - assert!(first_body.is_some()); - assert!(second_body.is_some()); assert_eq!( first_reads.load(AtomicOrdering::Relaxed), 0, @@ -6976,7 +7173,7 @@ mod tests { body_variant: rustfs_object_data_cache::ObjectDataCacheBodyVariant::FullObjectPlainV1, }); - let body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, reader, &info, @@ -6990,12 +7187,12 @@ mod tests { Some(Bytes::from_static(b"oops")), "test-bucket", "cached-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("size-mismatched buffered-body handoff should still return a response body"); let lookup = adapter.lookup_body(&plan).await; - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -7031,7 +7228,7 @@ mod tests { }) .expect("materialize-fill cache adapter should initialize"); - let first_body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _first_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, first_reader, &info, @@ -7045,11 +7242,12 @@ mod tests { None, "test-bucket", "materialized-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("materialize-fill handoff should succeed"); - let second_body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _second_body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, second_reader, &info, @@ -7063,12 +7261,11 @@ mod tests { None, "test-bucket", "materialized-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("follow-up cache hit should succeed"); - assert!(first_body.is_some()); - assert!(second_body.is_some()); assert_eq!( first_reads.load(AtomicOrdering::Relaxed), 2, @@ -7102,7 +7299,7 @@ mod tests { }) .expect("materialize-fill cache adapter should initialize"); - let body = DefaultObjectUsecase::build_get_object_body_with_cache( + let _body = DefaultObjectUsecase::build_get_object_body_with_cache( &adapter, reader, &info, @@ -7116,11 +7313,11 @@ mod tests { None, "test-bucket", "too-large-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("too-large cache candidate should use streaming fallback"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -7139,7 +7336,7 @@ mod tests { ..Default::default() }; - let body = DefaultObjectUsecase::build_get_object_body( + let _body = DefaultObjectUsecase::build_get_object_body( reader, &info, 4, @@ -7152,11 +7349,11 @@ mod tests { None, "test-bucket", "small-plain-object", + GetObjectBodyLifecycle::disabled(), ) .await .expect("build_get_object_body should keep small plain object on streaming path"); - assert!(body.is_some()); assert_eq!( reads.load(AtomicOrdering::Relaxed), 0, @@ -7877,6 +8074,7 @@ mod tests { 1, None, false, + GetObjectBodyLifecycle::disabled(), ) .await .expect("get object output context"); @@ -7907,6 +8105,31 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } + #[test] + fn parse_get_object_part_number_rejects_above_s3_max() { + let err = parse_part_number_i32_to_usize(Some(10001), "GET").expect_err("partNumber above S3 max must fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + assert_eq!(err.message(), Some("GET: partNumber must be between 1 and 10000")); + } + + #[test] + fn validate_get_object_part_number_rejects_missing_part() { + let info = ObjectInfo { + parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo { + number: 1, + ..Default::default() + }]), + ..Default::default() + }; + + let err = + DefaultObjectUsecase::validate_get_object_part_number(Some(2), &info).expect_err("missing requested part must fail"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidPart); + assert!(DefaultObjectUsecase::validate_get_object_part_number(Some(1), &info).is_ok()); + } + #[tokio::test] async fn execute_get_object_rejects_range_with_part_number() { let input = GetObjectInput::builder() diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index 5cbffa421..f7a045636 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -20,6 +20,7 @@ use crate::config::{RustFSBufferConfig, WorkloadProfile, is_buffer_profile_enabl use crate::error::ApiError; use crate::server::cors; use crate::storage::ecfs::ListObjectUnorderedQuery; +use crate::storage::storage_api::ecfs_extend_consumer::contract::multipart::MAX_MULTIPART_PART_NUMBER; use crate::storage::storage_api::ecfs_extend_consumer::contract::{ bucket::{BucketOperations, BucketOptions}, object::ObjectToDelete, @@ -1106,9 +1107,9 @@ pub(crate) async fn wrap_response_with_cors( pub(crate) fn parse_part_number_i32_to_usize(part_number: Option, op: &'static str) -> S3Result> { match part_number { None => Ok(None), - Some(n) if n <= 0 => Err(S3Error::with_message( + Some(n) if !(1..=MAX_MULTIPART_PART_NUMBER).contains(&n) => Err(S3Error::with_message( S3ErrorCode::InvalidArgument, - format!("{op}: invalid partNumber {n}, must be a positive integer"), + format!("{op}: partNumber must be between 1 and {MAX_MULTIPART_PART_NUMBER}"), )), Some(n) => Ok(Some(n as usize)), } diff --git a/rustfs/src/storage/s3_api/bucket.rs b/rustfs/src/storage/s3_api/bucket.rs index 230624831..1eb806955 100644 --- a/rustfs/src/storage/s3_api/bucket.rs +++ b/rustfs/src/storage/s3_api/bucket.rs @@ -562,6 +562,22 @@ mod tests { assert_eq!(output.next_marker.as_deref(), Some("dir%20a/key")); } + #[test] + fn test_list_objects_next_marker_uses_visible_common_prefix_with_internal_token() { + let marker = "asdf[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + let v2 = ListObjectsV2Output { + is_truncated: Some(true), + next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), + common_prefixes: Some(vec![CommonPrefix { + prefix: Some("asdf".to_string()), + }]), + ..Default::default() + }; + + let output = build_list_objects_output(v2, None); + assert_eq!(output.next_marker.as_deref(), Some("asdf")); + } + #[test] fn test_list_objects_next_marker_preserves_cache_like_key_suffix() { let marker = "key[rustfs_cache:v2]"; diff --git a/rustfs/src/storage/s3_api/multipart.rs b/rustfs/src/storage/s3_api/multipart.rs index 83aaeaf47..fe9352ef4 100644 --- a/rustfs/src/storage/s3_api/multipart.rs +++ b/rustfs/src/storage/s3_api/multipart.rs @@ -13,13 +13,14 @@ // limitations under the License. use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner}; -use crate::storage::storage_api::s3_api_consumer::multipart::contract::multipart::{ListMultipartsInfo, ListPartsInfo}; +use crate::storage::storage_api::s3_api_consumer::multipart::contract::multipart::{ + ListMultipartsInfo, ListPartsInfo, MAX_MULTIPART_PART_NUMBER, +}; use crate::storage::storage_api::s3_api_consumer::multipart::to_s3s_etag; use s3s::dto::{CommonPrefix, ListMultipartUploadsOutput, ListPartsOutput, MultipartUpload, Part, Timestamp}; use s3s::{S3Error, S3ErrorCode}; const MAX_MULTIPART_UPLOADS_LIST: i32 = 1000; -const MAX_MULTIPART_PART_NUMBER: i32 = 10000; #[derive(Debug, PartialEq, Eq)] pub(crate) struct ListPartsParams { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 7e3be8a34..4a66cca39 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -35,6 +35,7 @@ pub(crate) mod contract { } pub(crate) mod multipart { + pub(crate) const MAX_MULTIPART_PART_NUMBER: i32 = 10000; pub(crate) use super::super::storage_contracts::{ListMultipartsInfo, ListPartsInfo}; #[cfg(test)] pub(crate) use super::super::storage_contracts::{MultipartInfo, PartInfo}; @@ -128,6 +129,10 @@ pub(crate) mod ecfs_extend_consumer { pub(crate) mod object { pub(crate) use super::super::super::contract::object::ObjectToDelete; } + + pub(crate) mod multipart { + pub(crate) use super::super::super::contract::multipart::MAX_MULTIPART_PART_NUMBER; + } } pub(crate) type StorageObjectInfo = super::StorageObjectInfo; @@ -254,7 +259,9 @@ pub(crate) mod s3_api_consumer { pub(crate) mod contract { pub(crate) mod multipart { - pub(crate) use super::super::super::super::contract::multipart::{ListMultipartsInfo, ListPartsInfo}; + pub(crate) use super::super::super::super::contract::multipart::{ + ListMultipartsInfo, ListPartsInfo, MAX_MULTIPART_PART_NUMBER, + }; #[cfg(test)] pub(crate) use super::super::super::super::contract::multipart::{MultipartInfo, PartInfo}; diff --git a/scripts/s3-tests/run.sh b/scripts/s3-tests/run.sh index d9ca25c16..30b65c7c6 100755 --- a/scripts/s3-tests/run.sh +++ b/scripts/s3-tests/run.sh @@ -39,6 +39,13 @@ S3_REGION="${S3_REGION:-us-east-1}" S3_HOST="${S3_HOST:-127.0.0.1}" S3_PORT="${S3_PORT:-9000}" +# Keep the compatibility harness focused on foreground S3 API behavior. +# The background scanner can race short-lived test buckets and add avoidable +# metacache/listing pressure on small CI runners. +export RUSTFS_SCANNER_ENABLED="${RUSTFS_SCANNER_ENABLED:-false}" +export RUSTFS_SCANNER_START_DELAY_SECS="${RUSTFS_SCANNER_START_DELAY_SECS:-3600}" +export RUSTFS_SCANNER_CYCLE="${RUSTFS_SCANNER_CYCLE:-3600}" + # Test parameters TEST_MODE="${TEST_MODE:-single}" MAXFAIL="${MAXFAIL:-1}" @@ -259,6 +266,7 @@ Environment Variables: S3_ALT_ACCESS_KEY - Alt user access key (default: rustfsalt) S3_ALT_SECRET_KEY - Alt user secret key (default: rustfsalt) RUSTFS_SSE_S3_MASTER_KEY - Optional base64 32-byte key for local managed SSE fallback + RUSTFS_SCANNER_ENABLED - Enable background scanner for harness service (default: false) MAXFAIL - Stop after N failures, 0 = never stop (default: 1) XDIST - Enable parallel execution with N workers (default: 0) TEST_SCOPE - "implemented" (whitelist, default) or "all" (entire upstream suite) @@ -479,6 +487,9 @@ elif [ "${DEPLOY_MODE}" = "docker" ]; then -e RUSTFS_ACCESS_KEY="${S3_ACCESS_KEY}" \ -e RUSTFS_SECRET_KEY="${S3_SECRET_KEY}" \ -e RUSTFS_SSE_S3_MASTER_KEY="${RUSTFS_SSE_S3_MASTER_KEY}" \ + -e RUSTFS_SCANNER_ENABLED="${RUSTFS_SCANNER_ENABLED}" \ + -e RUSTFS_SCANNER_START_DELAY_SECS="${RUSTFS_SCANNER_START_DELAY_SECS}" \ + -e RUSTFS_SCANNER_CYCLE="${RUSTFS_SCANNER_CYCLE}" \ -e RUSTFS_VOLUMES="/data/rustfs0 /data/rustfs1 /data/rustfs2 /data/rustfs3" \ -v "/tmp/${CONTAINER_NAME}:/data" \ rustfs-ci || {