Compare commits

..

3 Commits

Author SHA1 Message Date
overtrue 3dc92c6dff test(e2e): update security suite membership 2026-08-23 06:18:36 +08:00
overtrue d299b82b21 test(e2e): require tier SSRF guard evidence 2026-08-23 05:40:27 +08:00
overtrue fd6046faf2 test(e2e): pin multipart part number limit 2026-08-23 05:33:25 +08:00
44 changed files with 340 additions and 3033 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
sha256-darwin=b8549d3362a69cca01c2a81f548bb06d5142d8a9ab4509487a656c8b3db1c164
sha256-linux=7ecd054965b4afa070af6deefdc37b5ca9f6a9b488dd5eef1ad0877378365b2f
+6 -3
View File
@@ -39,10 +39,11 @@ jobs:
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -88,10 +89,11 @@ jobs:
# either casing.
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -176,10 +178,11 @@ jobs:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout repository
- name: Checkout main branch
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
+1 -2
View File
@@ -30,8 +30,7 @@ make build-docker BUILD_OS=ubuntu22.04
- Crate membership: `Cargo.toml` `[workspace].members`
- Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
- Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
-2
View File
@@ -70,8 +70,6 @@ make pre-pr
> For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
### 🔒 Automated Pre-commit Hooks
#### What `make pre-commit` and `make pre-pr` actually run
Generated
+27 -22
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.4"
version = "1.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,6 +2522,12 @@ dependencies = [
"subtle",
]
[[package]]
name = "cty"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
@@ -5982,6 +5988,15 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
[[package]]
name = "libmimalloc-sys"
version = "0.1.49"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"cc",
"cty",
]
[[package]]
name = "libredox"
version = "0.1.20"
@@ -6382,6 +6397,14 @@ dependencies = [
"synstructure 0.13.2",
]
[[package]]
name = "mimalloc"
version = "0.1.52"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"libmimalloc-sys",
]
[[package]]
name = "mime"
version = "0.3.17"
@@ -9139,11 +9162,13 @@ dependencies = [
"insta",
"jiff",
"libc",
"libmimalloc-sys",
"libsystemd",
"matchit 0.9.2",
"md-5 0.11.0",
"metrics",
"metrics-util",
"mimalloc",
"mime_guess",
"opentelemetry",
"opentelemetry_sdk",
@@ -9179,8 +9204,6 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9852,24 +9875,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "rustfs-mimalloc"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
dependencies = [
"rustfs-mimalloc-sys",
]
[[package]]
name = "rustfs-mimalloc-sys"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
dependencies = [
"cc",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.3"
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+3 -676
View File
@@ -12,13 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use serde::{
Deserialize, Serialize,
de::{IgnoredAny, SeqAccess, Visitor},
ser::SerializeMap as _,
};
use serde::{Deserialize, Serialize, ser::SerializeMap as _};
use std::{
borrow::Cow,
collections::{HashMap, HashSet},
hash::{DefaultHasher, Hash, Hasher},
time::{Duration, SystemTime},
@@ -53,11 +48,6 @@ pub const DATA_USAGE_OBSERVED_OBJECT_NAME: &str = ".usage.observed.json";
// RUSTFS_COMPAT_TODO(scanner-usage-v2): keep .usage.json readable and removable during rolling upgrades from pre-v2 scanners. Remove after supported direct-upgrade sources all write .usage.v2.json.
pub const LEGACY_DATA_USAGE_OBJECT_NAME: &str = ".usage.json";
/// Fixed bucket for objects whose storage class is not in the scanner's
/// cycle-local tier registry. Keeping this key fixed prevents untrusted or
/// stale tier names from growing persisted per-tier maps without bound.
pub const UNKNOWN_TIER: &str = "UNKNOWN_TIER";
/// Returns true when `existing_last_update` is ahead of `now` by more than
/// [`USAGE_LAST_UPDATE_FUTURE_TOLERANCE`], i.e. the persisted timestamp cannot be
/// trusted for staleness comparisons and a fresh snapshot save must be allowed.
@@ -88,301 +78,12 @@ impl TierStats {
&& self.num_objects.checked_add(u.num_objects).is_some()
}
/// Add tier counters without allowing a counter to wrap.
pub fn checked_add(&self, u: &TierStats) -> Option<TierStats> {
Some(TierStats {
total_size: self.total_size.checked_add(u.total_size)?,
num_versions: self.num_versions.checked_add(u.num_versions)?,
num_objects: self.num_objects.checked_add(u.num_objects)?,
})
}
/// True when this tier contributed nothing, i.e. merging it is a no-op.
pub fn is_empty(&self) -> bool {
self.total_size == 0 && self.num_versions == 0 && self.num_objects == 0
}
}
/// Bounded diagnostics for objects whose tier is absent from the cycle
/// registry. Counters are authoritative; diagnostics are only a small,
/// redacted reconciliation aid and may be dropped at the configured caps.
pub const UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP: usize = 64;
pub const UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP: usize = 4096;
pub const UNKNOWN_TIER_DIAGNOSTIC_TTL: Duration = Duration::from_secs(60 * 60);
const UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES: usize = 256;
const UNKNOWN_TIER_DIAGNOSTIC_INPUT_CAP: usize = UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP * 4;
#[derive(Clone, Debug, Default, Serialize, PartialEq, Eq)]
pub struct UnknownTierStats {
/// Logical bytes retained in the scanner's normal usage total.
pub unknown_bytes: u64,
/// Physical bytes recorded in the per-tier accounting dimension.
///
/// Older writers only had `unknown_bytes`; decoding those snapshots keeps
/// this field at zero and the scanner fills it for new observations.
#[serde(default)]
pub unknown_physical_bytes: u64,
pub unknown_objects: u64,
pub unknown_versions: u64,
pub diagnostics_dropped: u64,
#[serde(default)]
pub diagnostics: Vec<String>,
#[serde(default)]
pub diagnostics_at: Option<SystemTime>,
/// A saturating update occurred; this snapshot cannot prove conservation.
/// The field is persisted so a restarted scanner cannot mistake a
/// saturated legacy aggregate for exact accounting evidence.
#[serde(default)]
pub counter_overflowed: bool,
}
#[derive(Default)]
struct BoundedDiagnostics {
entries: Vec<String>,
dropped: u64,
}
impl<'de> Deserialize<'de> for BoundedDiagnostics {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
struct BoundedDiagnosticsVisitor;
impl<'de> Visitor<'de> for BoundedDiagnosticsVisitor {
type Value = BoundedDiagnostics;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("a sequence of bounded tier diagnostics")
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: SeqAccess<'de>,
{
let mut bounded = BoundedDiagnostics::default();
let mut bytes = 0_usize;
let mut inspected = 0_usize;
loop {
if bounded.entries.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
|| bytes >= UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
|| inspected >= UNKNOWN_TIER_DIAGNOSTIC_INPUT_CAP
{
if sequence.next_element::<IgnoredAny>()?.is_none() {
break;
}
bounded.dropped = bounded.dropped.saturating_add(1);
continue;
}
let Some(diagnostic) = sequence.next_element::<Cow<'de, str>>()? else {
break;
};
inspected = inspected.saturating_add(1);
if !is_redacted_tier_diagnostic(&diagnostic)
|| bytes.saturating_add(diagnostic.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
|| bounded.entries.iter().any(|entry| entry == &diagnostic)
{
bounded.dropped = bounded.dropped.saturating_add(1);
continue;
}
bytes = bytes.saturating_add(diagnostic.len());
bounded.entries.push(diagnostic.into_owned());
}
Ok(bounded)
}
}
deserializer.deserialize_seq(BoundedDiagnosticsVisitor)
}
}
#[derive(Deserialize)]
struct UnknownTierStatsWire {
#[serde(default)]
unknown_bytes: u64,
#[serde(default)]
unknown_physical_bytes: u64,
#[serde(default)]
unknown_objects: u64,
#[serde(default)]
unknown_versions: u64,
#[serde(default)]
diagnostics_dropped: u64,
#[serde(default)]
diagnostics: BoundedDiagnostics,
#[serde(default)]
diagnostics_at: Option<SystemTime>,
#[serde(default)]
counter_overflowed: bool,
}
fn is_redacted_tier_diagnostic(diagnostic: &str) -> bool {
diagnostic
.strip_prefix("tier-hash:")
.is_some_and(|digest| digest.len() == 16 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()))
}
fn diagnostics_expired(at: SystemTime, now: SystemTime) -> bool {
now.duration_since(at).map_or(true, |age| age > UNKNOWN_TIER_DIAGNOSTIC_TTL)
}
fn checked_saturating_add(left: u64, right: u64, overflowed: &mut bool) -> u64 {
match left.checked_add(right) {
Some(value) => value,
None => {
*overflowed = true;
u64::MAX
}
}
}
impl<'de> Deserialize<'de> for UnknownTierStats {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let wire = UnknownTierStatsWire::deserialize(deserializer)?;
let mut stats = Self {
unknown_bytes: wire.unknown_bytes,
unknown_physical_bytes: wire.unknown_physical_bytes,
unknown_objects: wire.unknown_objects,
unknown_versions: wire.unknown_versions,
counter_overflowed: wire.counter_overflowed,
diagnostics_dropped: wire.diagnostics_dropped,
diagnostics: wire.diagnostics.entries,
diagnostics_at: wire.diagnostics_at,
};
stats.diagnostics_dropped =
checked_saturating_add(stats.diagnostics_dropped, wire.diagnostics.dropped, &mut stats.counter_overflowed);
if stats
.diagnostics_at
.is_some_and(|at| diagnostics_expired(at, SystemTime::now()))
{
stats.diagnostics.clear();
stats.diagnostics_at = None;
}
Ok(stats)
}
}
impl UnknownTierStats {
/// Record one observation where the logical and physical dimensions are
/// the same. Kept as a small compatibility helper for callers that only
/// have one size value.
pub fn record(&mut self, tier: &str, bytes: u64, versions: u64, objects: u64) {
self.record_dimensions(tier, bytes, bytes, versions, objects);
}
/// Record one observation without conflating logical usage with physical
/// tier bytes. Both counters are saturating so malformed metadata cannot
/// wrap an aggregate.
pub fn record_dimensions(&mut self, tier: &str, logical_bytes: u64, physical_bytes: u64, versions: u64, objects: u64) {
self.unknown_bytes = checked_saturating_add(self.unknown_bytes, logical_bytes, &mut self.counter_overflowed);
self.unknown_physical_bytes =
checked_saturating_add(self.unknown_physical_bytes, physical_bytes, &mut self.counter_overflowed);
self.unknown_objects = checked_saturating_add(self.unknown_objects, objects, &mut self.counter_overflowed);
self.unknown_versions = checked_saturating_add(self.unknown_versions, versions, &mut self.counter_overflowed);
let digest = {
let mut hasher = DefaultHasher::new();
// Bound hashing work for hostile metadata while retaining enough
// length/prefix entropy to reconcile repeated observations.
hasher.write_u64(u64::try_from(tier.len()).unwrap_or(u64::MAX));
let bounded = &tier.as_bytes()[..tier.len().min(UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES)];
hasher.write(bounded);
hasher.write_u8(u8::from(bounded.iter().any(|byte| byte.is_ascii_control())));
format!("tier-hash:{:016x}", hasher.finish())
};
let now = SystemTime::now();
if self.diagnostics_at.is_some_and(|at| diagnostics_expired(at, now)) {
self.diagnostics.clear();
}
self.diagnostics_at = Some(now);
if self.diagnostics.iter().any(|entry| entry == &digest) {
return;
}
let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
|| current_bytes.saturating_add(digest.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
{
self.diagnostics_dropped = checked_saturating_add(1, self.diagnostics_dropped, &mut self.counter_overflowed);
return;
}
self.diagnostics.push(digest);
}
pub fn merge(&mut self, other: &Self) {
self.unknown_bytes = checked_saturating_add(self.unknown_bytes, other.unknown_bytes, &mut self.counter_overflowed);
self.unknown_physical_bytes =
checked_saturating_add(self.unknown_physical_bytes, other.unknown_physical_bytes, &mut self.counter_overflowed);
self.unknown_objects = checked_saturating_add(self.unknown_objects, other.unknown_objects, &mut self.counter_overflowed);
self.unknown_versions =
checked_saturating_add(self.unknown_versions, other.unknown_versions, &mut self.counter_overflowed);
self.diagnostics_dropped =
checked_saturating_add(self.diagnostics_dropped, other.diagnostics_dropped, &mut self.counter_overflowed);
self.counter_overflowed |= other.counter_overflowed;
let now = SystemTime::now();
if self.diagnostics_at.is_some_and(|at| diagnostics_expired(at, now)) {
self.diagnostics.clear();
self.diagnostics_at = None;
}
if !other.diagnostics_at.is_some_and(|at| diagnostics_expired(at, now)) {
for diagnostic in &other.diagnostics {
if !is_redacted_tier_diagnostic(diagnostic) {
self.diagnostics_dropped = checked_saturating_add(1, self.diagnostics_dropped, &mut self.counter_overflowed);
continue;
}
if self.diagnostics.iter().any(|entry| entry == diagnostic) {
continue;
}
let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
|| current_bytes.saturating_add(diagnostic.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
{
self.diagnostics_dropped = checked_saturating_add(1, self.diagnostics_dropped, &mut self.counter_overflowed);
continue;
}
self.diagnostics.push(diagnostic.clone());
}
}
if !self.diagnostics.is_empty() {
self.diagnostics_at = Some(now);
}
}
pub fn fits_add(&self, other: &Self) -> bool {
!self.counter_overflowed
&& !other.counter_overflowed
&& self.unknown_bytes.checked_add(other.unknown_bytes).is_some()
&& self
.unknown_physical_bytes
.checked_add(other.unknown_physical_bytes)
.is_some()
&& self.unknown_objects.checked_add(other.unknown_objects).is_some()
&& self.unknown_versions.checked_add(other.unknown_versions).is_some()
&& self.diagnostics_dropped.checked_add(other.diagnostics_dropped).is_some()
}
pub fn checked_add(&self, other: &Self) -> Option<Self> {
if !self.fits_add(other) {
return None;
}
let mut merged = self.clone();
merged.merge(other);
Some(merged)
}
pub fn is_empty(&self) -> bool {
!self.counter_overflowed
&& self.unknown_bytes == 0
&& self.unknown_physical_bytes == 0
&& self.unknown_objects == 0
&& self.unknown_versions == 0
&& self.diagnostics_dropped == 0
&& self.diagnostics.is_empty()
}
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct AllTierStats {
pub tiers: HashMap<String, TierStats>,
@@ -423,31 +124,6 @@ impl AllTierStats {
.iter()
.all(|(tier, right)| self.tiers.get(tier).is_none_or(|left| left.fits_add(right)))
}
/// Fold keys from an older cache that are no longer present in the
/// current registry into the fixed unknown bucket. Built-in storage
/// classes remain known even when no remote tier is configured.
pub fn fold_unknown_tiers<'a, I>(&mut self, known_tiers: I)
where
I: IntoIterator<Item = &'a str>,
{
let known: HashSet<&str> = known_tiers.into_iter().collect();
let mut unknown = self.tiers.remove(UNKNOWN_TIER).unwrap_or_default();
let retired_tiers: Vec<String> = self
.tiers
.keys()
.filter(|tier| tier.as_str() != "STANDARD" && tier.as_str() != "REDUCED_REDUNDANCY" && !known.contains(tier.as_str()))
.cloned()
.collect();
for tier in retired_tiers {
if let Some(stats) = self.tiers.remove(&tier) {
unknown = unknown.add(&stats);
}
}
if !unknown.is_empty() {
self.tiers.insert(UNKNOWN_TIER.to_string(), unknown);
}
}
}
/// Bucket target usage info provides replication statistics
@@ -535,10 +211,6 @@ pub struct DataUsageInfo {
/// tier exists, so an absent value means "not accounted", never "zero".
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tier_stats: Option<AllTierStats>,
/// Bounded diagnostics and separate logical/physical counters for objects
/// classified into [`UNKNOWN_TIER`].
#[serde(default, skip_serializing_if = "Option::is_none")]
pub unknown_tier_stats: Option<UnknownTierStats>,
/// Total number of buckets in this cluster
pub buckets_count: u64,
@@ -635,48 +307,6 @@ pub struct DiskUsageStatus {
pub snapshot_exists: bool,
}
/// Independent conservation evidence for a scanner summary.
///
/// The totals are maintained while objects are accounted and are deliberately
/// not reconstructed from the tier map at publish time. `*_known` records the
/// portion represented by the corresponding accounting dimension.
#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub struct TierAccountingProof {
pub logical_total: u64,
pub logical_known: u64,
pub physical_total: u64,
pub physical_known: u64,
#[serde(default)]
pub overflowed: bool,
}
impl TierAccountingProof {
pub fn checked_add(self, other: Self) -> Option<Self> {
if self.overflowed || other.overflowed {
return None;
}
Some(Self {
logical_total: self.logical_total.checked_add(other.logical_total)?,
logical_known: self.logical_known.checked_add(other.logical_known)?,
physical_total: self.physical_total.checked_add(other.physical_total)?,
physical_known: self.physical_known.checked_add(other.physical_known)?,
overflowed: false,
})
}
pub fn saturating_add(&mut self, other: Self) {
let Some(merged) = (*self).checked_add(other) else {
self.logical_total = self.logical_total.saturating_add(other.logical_total);
self.logical_known = self.logical_known.saturating_add(other.logical_known);
self.physical_total = self.physical_total.saturating_add(other.physical_total);
self.physical_known = self.physical_known.saturating_add(other.physical_known);
self.overflowed = true;
return;
};
*self = merged;
}
}
/// Size summary for a single object or group of objects
#[derive(Debug, Default, Clone)]
pub struct SizeSummary {
@@ -706,10 +336,6 @@ pub struct SizeSummary {
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
/// Per-tier accounting, keyed by storage class or remote tier name
pub tier_stats: HashMap<String, TierStats>,
/// Counters and bounded diagnostics for unknown tiers in this summary.
pub unknown_tier_stats: UnknownTierStats,
/// Independent logical/physical conservation evidence.
pub tier_accounting_proof: TierAccountingProof,
}
/// Replication target size summary
@@ -801,10 +427,6 @@ impl<'de> Deserialize<'de> for SizeHistogram {
}
impl SizeHistogram {
pub fn is_empty(&self) -> bool {
self.0.iter().all(|value| *value == 0)
}
pub fn add(&mut self, size: u64) {
let intervals = [
(0, 1024 - 1), // LESS_THAN_1024_B
@@ -919,10 +541,6 @@ impl<'de> Deserialize<'de> for VersionsHistogram {
}
impl VersionsHistogram {
pub fn is_empty(&self) -> bool {
self.0.iter().all(|value| *value == 0)
}
pub fn add(&mut self, count: u64) {
let intervals = [
(0, 0), // UNVERSIONED
@@ -1063,13 +681,6 @@ pub struct DataUsageEntry {
/// observed tier-classified objects.
#[serde(default)]
pub all_tier_stats: Option<AllTierStats>,
/// Bounded unknown-tier reconciliation state for this cache entry.
#[serde(default)]
pub unknown_tier_stats: Option<UnknownTierStats>,
/// Optional conservation proof. Missing values are legacy/unproven, not
/// zero-valued evidence.
#[serde(default)]
pub tier_accounting_proof: Option<TierAccountingProof>,
}
impl Serialize for DataUsageEntry {
@@ -1080,9 +691,7 @@ impl Serialize for DataUsageEntry {
// Keep entries map-encoded so older readers can ignore fields appended
// by newer scanner versions during rolling upgrades. The derived
// (array) encoding made any appended field a decode error for them.
let mut state = serializer.serialize_map(Some(
11 + usize::from(self.unknown_tier_stats.is_some()) + usize::from(self.tier_accounting_proof.is_some()),
))?;
let mut state = serializer.serialize_map(Some(11))?;
state.serialize_entry("children", &self.children)?;
state.serialize_entry("size", &self.size)?;
state.serialize_entry("objects", &self.objects)?;
@@ -1094,35 +703,11 @@ impl Serialize for DataUsageEntry {
state.serialize_entry("compacted", &self.compacted)?;
state.serialize_entry("failed_objects", &self.failed_objects)?;
state.serialize_entry("all_tier_stats", &self.all_tier_stats)?;
// Keep the legacy no-unknown shape byte-for-byte stable. Once unknown
// accounting exists, append its map field before the optional proof.
if let Some(unknown_tier_stats) = self.unknown_tier_stats.as_ref() {
state.serialize_entry("unknown_tier_stats", unknown_tier_stats)?;
}
if let Some(proof) = self.tier_accounting_proof.as_ref() {
state.serialize_entry("tier_accounting_proof", proof)?;
}
state.end()
}
}
impl DataUsageEntry {
fn has_local_usage(&self) -> bool {
self.size != 0
|| self.objects != 0
|| self.versions != 0
|| self.delete_markers != 0
|| self.failed_objects != 0
|| self.obj_sizes.0.iter().any(|value| *value != 0)
|| self.obj_versions.0.iter().any(|value| *value != 0)
|| self.replication_stats.as_ref().is_some_and(|stats| !stats.is_empty())
|| self
.all_tier_stats
.as_ref()
.is_some_and(|stats| stats.tiers.values().any(|tier| !tier.is_empty()))
|| self.unknown_tier_stats.as_ref().is_some_and(|stats| !stats.is_empty())
}
pub fn add_child(&mut self, hash: &DataUsageHash) {
if self.children.contains(&hash.key()) {
return;
@@ -1131,8 +716,6 @@ impl DataUsageEntry {
}
pub fn merge(&mut self, other: &DataUsageEntry) {
let self_had_local_usage = self.has_local_usage();
let other_had_local_usage = other.has_local_usage();
self.objects += other.objects;
self.versions += other.versions;
self.delete_markers += other.delete_markers;
@@ -1161,29 +744,6 @@ impl DataUsageEntry {
if let Some(o_tiers) = other.all_tier_stats.as_ref().filter(|tiers| !tiers.is_empty()) {
self.all_tier_stats.get_or_insert_with(AllTierStats::new).merge(o_tiers);
}
if let Some(other_unknown) = other.unknown_tier_stats.as_ref() {
self.unknown_tier_stats
.get_or_insert_with(UnknownTierStats::default)
.merge(other_unknown);
}
self.tier_accounting_proof = match (self.tier_accounting_proof, other.tier_accounting_proof) {
(Some(mut left), Some(right)) => {
left.saturating_add(right);
Some(left)
}
(None, Some(right)) if !self_had_local_usage => Some(right),
(Some(left), None) if !other_had_local_usage => Some(left),
(None, None) => None,
_ => None,
};
// A saturated unknown-tier counter invalidates conservation evidence;
// never let an otherwise valid proof hide that loss of precision.
if self.unknown_tier_stats.as_ref().is_some_and(|stats| stats.counter_overflowed)
&& let Some(proof) = self.tier_accounting_proof.as_mut()
{
proof.overflowed = true;
}
self.obj_sizes.merge_from(&other.obj_sizes);
self.obj_versions.merge_from(&other.obj_versions);
@@ -1197,15 +757,6 @@ impl DataUsageEntry {
self.all_tier_stats.get_or_insert_with(AllTierStats::new).add_sizes(tiers);
}
pub fn add_unknown_tier_stats(&mut self, stats: &UnknownTierStats) {
if stats.is_empty() {
return;
}
self.unknown_tier_stats
.get_or_insert_with(UnknownTierStats::default)
.merge(stats);
}
pub fn checked_merge(&mut self, other: &DataUsageEntry) -> bool {
let scalar_counts_fit = self.objects.checked_add(other.objects).is_some()
&& self.versions.checked_add(other.versions).is_some()
@@ -1269,21 +820,8 @@ impl DataUsageEntry {
(_, None) | (None, Some(_)) => true,
(Some(left), Some(right)) => left.fits_merge(right),
};
let unknown_tier_stats_fit = match (&self.unknown_tier_stats, &other.unknown_tier_stats) {
(None, None) => true,
(Some(left), None) => !left.counter_overflowed,
(None, Some(right)) => !right.counter_overflowed,
(Some(left), Some(right)) => left.fits_add(right),
};
let proof_fit = match (self.tier_accounting_proof, other.tier_accounting_proof) {
(Some(left), Some(right)) => left.checked_add(right).is_some(),
(Some(proof), None) | (None, Some(proof)) => !proof.overflowed,
(None, None) => true,
};
if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit || !unknown_tier_stats_fit || !proof_fit
{
if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit {
return false;
}
self.merge(other);
@@ -1801,7 +1339,6 @@ impl DataUsageCache {
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: flat.unknown_tier_stats.filter(|stats| !stats.is_empty()),
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
buckets_usage,
usage_snapshot_complete: self.info.snapshot_complete,
@@ -2229,20 +1766,6 @@ impl SizeSummary {
self.replica_count = self.replica_count.saturating_add(other.replica_count);
self.pending_count = self.pending_count.saturating_add(other.pending_count);
self.failed_count = self.failed_count.saturating_add(other.failed_count);
self.unknown_tier_stats.merge(&other.unknown_tier_stats);
self.tier_accounting_proof.saturating_add(other.tier_accounting_proof);
if self.unknown_tier_stats.counter_overflowed {
self.tier_accounting_proof.overflowed = true;
}
// A disk/bucket aggregate is assembled from many object summaries.
// Keep the per-tier dimension in lockstep with the scalar counters;
// dropping this map here would recreate the original silent-loss bug
// at the cross-disk merge boundary.
for (tier, stats) in &other.tier_stats {
let entry = self.tier_stats.entry(tier.clone()).or_default();
*entry = entry.add(stats);
}
// Merge replication target stats
for (target, stats) in &other.repl_target_stats {
@@ -2355,59 +1878,6 @@ mod tests {
);
}
#[test]
fn retired_tier_stats_fold_into_fixed_unknown_bucket() {
let mut stats = AllTierStats::default();
stats.tiers.insert(
"RETIRED".to_string(),
TierStats {
total_size: 9,
num_versions: 2,
num_objects: 1,
},
);
stats.tiers.insert(
"WARM".to_string(),
TierStats {
total_size: 4,
num_versions: 1,
num_objects: 1,
},
);
stats.fold_unknown_tiers(["WARM"]);
assert!(!stats.tiers.contains_key("RETIRED"));
assert_eq!(stats.tiers.get("WARM").map(|v| v.total_size), Some(4));
assert_eq!(stats.tiers.get(UNKNOWN_TIER).map(|v| v.total_size), Some(9));
}
#[test]
fn unknown_tier_stats_deserialization_keeps_diagnostics_bounded() {
#[derive(Serialize)]
struct RawUnknownTierStats {
unknown_bytes: u64,
diagnostics: Vec<String>,
diagnostics_dropped: u64,
}
let mut diagnostics = vec!["tier-hash:0123456789abcdef".to_string(); UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP + 8];
diagnostics.push("raw-tier-name-that-must-not-be-exposed".to_string());
diagnostics.push("x".repeat(UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP + 1));
let encoded = rmp_serde::to_vec_named(&RawUnknownTierStats {
unknown_bytes: 7,
diagnostics,
diagnostics_dropped: 3,
})
.expect("unknown tier stats should encode");
let decoded: UnknownTierStats = rmp_serde::from_slice(&encoded).expect("unknown tier stats should decode");
assert_eq!(decoded.unknown_bytes, 7);
assert_eq!(decoded.diagnostics.len(), 1);
assert!(decoded.diagnostics_dropped >= 3);
assert!(decoded.diagnostics.iter().all(|entry| is_redacted_tier_diagnostic(entry)));
}
#[test]
fn checked_merge_rejects_overflowing_tier_totals() {
let mut left = tier_entry(
@@ -2431,57 +1901,6 @@ mod tests {
assert_eq!(left.all_tier_stats.expect("left is untouched").tiers["WARM"].total_size, u64::MAX);
}
#[test]
fn checked_merge_rejects_unknown_counter_overflow_on_either_side() {
let overflowed = DataUsageEntry {
unknown_tier_stats: Some(UnknownTierStats {
counter_overflowed: true,
..Default::default()
}),
tier_accounting_proof: Some(TierAccountingProof {
logical_total: 1,
logical_known: 1,
..Default::default()
}),
..Default::default()
};
let mut left = DataUsageEntry::default();
assert!(!left.checked_merge(&overflowed));
let mut left = overflowed.clone();
assert!(!left.checked_merge(&DataUsageEntry::default()));
}
#[test]
fn checked_merge_rejects_one_sided_overflowed_proof_without_mutation() {
let mut left = DataUsageEntry {
size: 1,
tier_accounting_proof: Some(TierAccountingProof {
logical_total: 1,
logical_known: 1,
overflowed: true,
..Default::default()
}),
..Default::default()
};
let legacy = DataUsageEntry {
size: 2,
..Default::default()
};
assert!(!left.checked_merge(&legacy));
assert_eq!(left.size, 1);
assert_eq!(
left.tier_accounting_proof,
Some(TierAccountingProof {
logical_total: 1,
logical_known: 1,
overflowed: true,
..Default::default()
})
);
}
/// Entry shape released before per-tier accounting, using the derived
/// (array) encoding those writers produced.
#[derive(Serialize, Deserialize)]
@@ -2521,72 +1940,6 @@ mod tests {
assert_eq!(legacy.objects, 0);
}
#[test]
fn tier_accounting_proof_is_optional_and_map_encoded() {
let entry = DataUsageEntry {
tier_accounting_proof: Some(TierAccountingProof {
logical_total: 11,
logical_known: 11,
physical_total: 7,
physical_known: 7,
..Default::default()
}),
..Default::default()
};
let encoded = rmp_serde::to_vec(&entry).expect("proof-bearing entry should encode");
let decoded: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("proof-bearing entry should decode");
assert_eq!(decoded.tier_accounting_proof, entry.tier_accounting_proof);
let legacy = LegacyEntry {
children: DataUsageHashMap::default(),
size: 12,
objects: 3,
versions: 4,
delete_markers: 1,
obj_sizes: SizeHistogram::default(),
obj_versions: VersionsHistogram::default(),
replication_stats: None,
compacted: false,
failed_objects: 2,
};
let legacy_bytes = rmp_serde::to_vec(&legacy).expect("legacy entry should encode");
let decoded: DataUsageEntry = rmp_serde::from_slice(&legacy_bytes).expect("legacy entry should decode");
assert!(decoded.tier_accounting_proof.is_none());
}
#[test]
fn empty_entry_merge_preserves_accounting_proof() {
let mut entry = DataUsageEntry {
tier_accounting_proof: Some(TierAccountingProof {
logical_total: 7,
logical_known: 7,
physical_total: 7,
physical_known: 7,
..Default::default()
}),
..Default::default()
};
entry.merge(&DataUsageEntry::default());
assert!(entry.tier_accounting_proof.is_some());
let mut empty = DataUsageEntry::default();
empty.merge(&entry);
assert_eq!(empty.tier_accounting_proof, entry.tier_accounting_proof);
}
#[test]
fn unknown_counter_overflow_is_not_empty_and_survives_roundtrip() {
let stats = UnknownTierStats {
counter_overflowed: true,
..Default::default()
};
assert!(!stats.is_empty());
let encoded = rmp_serde::to_vec_named(&stats).expect("overflow marker should encode");
let decoded: UnknownTierStats = rmp_serde::from_slice(&encoded).expect("overflow marker should decode");
assert!(decoded.counter_overflowed);
assert!(!decoded.is_empty());
}
#[test]
fn legacy_array_encoded_entries_still_load() {
let legacy = LegacyEntry {
@@ -3355,32 +2708,6 @@ mod tests {
);
}
#[test]
fn test_dui_filters_empty_unknown_tier_usage_from_the_flattened_tree() {
let root_hash = hash_path("root");
let bucket_hash = hash_path("bucket-a");
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: "root".to_string(),
..Default::default()
},
..Default::default()
};
cache.replace_hashed(&root_hash, &None, &DataUsageEntry::default());
let child = DataUsageEntry {
objects: 1,
unknown_tier_stats: Some(UnknownTierStats::default()),
..Default::default()
};
cache.replace_hashed(&bucket_hash, &Some(root_hash), &child);
let info = cache.dui("root", &["bucket-a".to_string()]);
assert_eq!(info.objects_total_count, 1);
assert!(info.unknown_tier_stats.is_none());
}
#[test]
fn test_data_usage_entry_merge_preserves_replication_targets() {
let mut base = DataUsageEntry {
+33 -23
View File
@@ -17,14 +17,14 @@
//! These tests verify that RustFS properly enforces security-sensitive
//! controls by issuing real requests against a running server and asserting
//! the concrete outcome of each control:
//! - DoS protection (oversized tagging payloads, excessive multipart parts)
//! - DoS protection (oversized tagging payloads, out-of-range multipart part numbers)
//! - SSRF prevention (internal/private endpoints rejected for tiering)
//! - Race condition handling (concurrent writes converge without corruption)
use crate::common::{RustFSTestEnvironment, awscurl_available, awscurl_put, init_logging};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Tag, Tagging};
use aws_sdk_s3::types::{Tag, Tagging};
use std::error::Error;
use tracing::info;
@@ -88,9 +88,9 @@ async fn test_large_xml_body_rejection() -> Result<(), Box<dyn Error + Send + Sy
Ok(())
}
/// Excessive multipart parts must be rejected.
/// Multipart part numbers above the S3 limit must be rejected.
#[tokio::test]
async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + Sync>> {
async fn test_multipart_part_number_above_limit() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
@@ -108,18 +108,23 @@ async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + S
let upload_id = create_result.upload_id().expect("upload_id should be present").to_string();
// Try to complete with too many parts (should be rejected).
let mut parts = Vec::new();
for i in 1..=10001 {
parts.push(CompletedPart::builder().part_number(i).e_tag(format!("etag-{i}")).build());
}
let result = client
.complete_multipart_upload()
client
.upload_part()
.bucket(&bucket_name)
.key("test-large")
.upload_id(&upload_id)
.multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(parts)).build())
.part_number(10000)
.body(ByteStream::from_static(b"upper-bound part"))
.send()
.await?;
let result = client
.upload_part()
.bucket(&bucket_name)
.key("test-large")
.upload_id(&upload_id)
.part_number(10001)
.body(ByteStream::from_static(b"out-of-range part"))
.send()
.await;
@@ -133,7 +138,13 @@ async fn test_excessive_multipart_parts() -> Result<(), Box<dyn Error + Send + S
.await;
let _ = client.delete_bucket().bucket(&bucket_name).send().await;
assert!(result.is_err(), "Server should reject excessive multipart parts");
let err = result.expect_err("server must reject excessive multipart parts");
let code = err.as_service_error().and_then(ProvideErrorMetadata::code);
assert_eq!(
code,
Some("InvalidArgument"),
"Part number 10001 should be rejected with InvalidArgument, got code {code:?}, err: {err:?}"
);
env.stop_server();
Ok(())
@@ -217,12 +228,8 @@ async fn test_concurrent_object_operations() -> Result<(), Box<dyn Error + Send
/// Internal/private endpoints must be rejected as remote tier backends (SSRF).
///
/// This issues a real admin AddTier call (`PUT /rustfs/admin/v3/tier`) for each
/// internal/private endpoint and asserts the server rejects it (non-2xx, so the
/// signed request helper returns an error). An internal endpoint must never be
/// accepted as a tier backend. The rejection may originate from explicit
/// SSRF/internal-address filtering or from the backend connectivity/credential
/// validation performed during AddTier; either way the security-relevant
/// outcome — the internal endpoint is not accepted — is asserted here.
/// internal/private endpoint and asserts the request reaches the outbound URL
/// guard. Connectivity or credential failures do not prove SSRF protection.
///
/// The admin API is exercised via signed `awscurl` requests, matching the
/// pattern used by the other admin-API E2E tests in this crate; the test is
@@ -263,10 +270,13 @@ async fn test_tiering_url_validation() -> Result<(), Box<dyn Error + Send + Sync
})
.to_string();
let result = awscurl_put(&tier_url, &body, &env.access_key, &env.secret_key).await;
let err = awscurl_put(&tier_url, &body, &env.access_key, &env.secret_key)
.await
.expect_err("AddTier must reject internal endpoints");
let rendered = err.to_string();
assert!(
result.is_err(),
"AddTier must reject internal endpoint {endpoint}, but it was accepted: {result:?}"
rendered.contains("TierAddFailed") && rendered.contains("tier endpoint is not allowed"),
"AddTier rejected {endpoint} outside the outbound URL guard: {rendered}"
);
}
+5 -7
View File
@@ -456,13 +456,11 @@ pub mod rpc {
ScannerBucketListing, ScannerPeerActivity, TONIC_RPC_PREFIX, TonicInterceptor, build_put_file_auth_trailer,
check_and_record_signed_rpc_nonce, gen_signature_headers, gen_tonic_replay_scope_headers, gen_tonic_signature_headers,
gen_tonic_signature_interceptor, node_service_time_out_client, node_service_time_out_client_no_auth,
normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability,
sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_ns_scanner_capability, verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer,
verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
verify_tonic_rpc_signature_with_bootstrap,
normalize_tonic_rpc_audience, set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_put_file_capability,
sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_put_file_capability, verify_rpc_signature,
verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
verify_tonic_rpc_response_proof, verify_tonic_rpc_signature, verify_tonic_rpc_signature_with_bootstrap,
};
}
+13 -68
View File
@@ -94,7 +94,6 @@ const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13;
const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096;
const REPLAY_CACHE_AUTO_MAX_CAPACITY: usize = 33_554_432;
const NS_SCANNER_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-capability-v3";
const NS_SCANNER_TIER_REGISTRY_GENERATION_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-tier-registry-generation-v1";
pub const TONIC_RPC_PREFIX: &str = "/node_service.NodeService";
static INTERNODE_RPC_SIGNATURE_STRICT: LazyLock<bool> = LazyLock::new(|| {
get_env_bool(
@@ -637,79 +636,40 @@ pub fn verify_put_file_capability(challenge: Uuid, server_epoch: Uuid, version:
.map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid put_file capability proof"))
}
fn update_ns_scanner_capability_mac(
mac: &mut HmacSha256,
challenge: Uuid,
server_epoch: Uuid,
supports_tier_registry_generation: bool,
) {
fn update_ns_scanner_capability_mac(mac: &mut HmacSha256, challenge: Uuid, server_epoch: Uuid) {
mac.update(NS_SCANNER_CAPABILITY_AUTH_DOMAIN);
mac.update(&NS_SCANNER_PROTOCOL_VERSION.to_be_bytes());
mac.update(challenge.as_bytes());
mac.update(server_epoch.as_bytes());
if supports_tier_registry_generation {
// The optional response capability is part of the authenticated
// scope. A proxy cannot turn an old/unsupported peer into a worker
// that receives generation-fenced scanner work.
mac.update(NS_SCANNER_TIER_REGISTRY_GENERATION_AUTH_DOMAIN);
}
}
fn generate_ns_scanner_capability_proof(
secret: &str,
challenge: Uuid,
server_epoch: Uuid,
supports_tier_registry_generation: bool,
) -> std::io::Result<Vec<u8>> {
fn generate_ns_scanner_capability_proof(secret: &str, challenge: Uuid, server_epoch: Uuid) -> std::io::Result<Vec<u8>> {
if challenge.is_nil() || server_epoch.is_nil() {
return Err(std::io::Error::other("Invalid namespace scanner capability scope"));
}
let mut mac =
<HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?;
update_ns_scanner_capability_mac(&mut mac, challenge, server_epoch, supports_tier_registry_generation);
update_ns_scanner_capability_mac(&mut mac, challenge, server_epoch);
Ok(mac.finalize().into_bytes().to_vec())
}
fn verify_ns_scanner_capability_proof(
secret: &str,
challenge: Uuid,
server_epoch: Uuid,
proof: &[u8],
supports_tier_registry_generation: bool,
) -> std::io::Result<()> {
fn verify_ns_scanner_capability_proof(secret: &str, challenge: Uuid, server_epoch: Uuid, proof: &[u8]) -> std::io::Result<()> {
if challenge.is_nil() || server_epoch.is_nil() {
return Err(std::io::Error::other("Invalid namespace scanner capability scope"));
}
let mut mac =
<HmacSha256 as KeyInit>::new_from_slice(secret.as_bytes()).map_err(|_| std::io::Error::other("Invalid RPC HMAC key"))?;
update_ns_scanner_capability_mac(&mut mac, challenge, server_epoch, supports_tier_registry_generation);
update_ns_scanner_capability_mac(&mut mac, challenge, server_epoch);
mac.verify_slice(proof)
.map_err(|_| std::io::Error::new(std::io::ErrorKind::PermissionDenied, "Invalid namespace scanner capability proof"))
}
pub fn sign_ns_scanner_capability(challenge: Uuid, server_epoch: Uuid) -> std::io::Result<Vec<u8>> {
sign_ns_scanner_capability_with_tier_registry_generation(challenge, server_epoch, false)
generate_ns_scanner_capability_proof(&get_shared_secret()?, challenge, server_epoch)
}
pub fn verify_ns_scanner_capability(challenge: Uuid, server_epoch: Uuid, proof: &[u8]) -> std::io::Result<()> {
verify_ns_scanner_capability_with_tier_registry_generation(challenge, server_epoch, proof, false)
}
pub fn sign_ns_scanner_capability_with_tier_registry_generation(
challenge: Uuid,
server_epoch: Uuid,
supports_tier_registry_generation: bool,
) -> std::io::Result<Vec<u8>> {
generate_ns_scanner_capability_proof(&get_shared_secret()?, challenge, server_epoch, supports_tier_registry_generation)
}
pub fn verify_ns_scanner_capability_with_tier_registry_generation(
challenge: Uuid,
server_epoch: Uuid,
proof: &[u8],
supports_tier_registry_generation: bool,
) -> std::io::Result<()> {
verify_ns_scanner_capability_proof(&get_shared_secret()?, challenge, server_epoch, proof, supports_tier_registry_generation)
verify_ns_scanner_capability_proof(&get_shared_secret()?, challenge, server_epoch, proof)
}
#[derive(Clone, Copy)]
@@ -1749,28 +1709,13 @@ mod tests {
let secret = "test-scanner-capability-secret";
let challenge = Uuid::new_v4();
let server_epoch = Uuid::new_v4();
let proof = generate_ns_scanner_capability_proof(secret, challenge, server_epoch, false)
.expect("capability proof should be generated");
let proof =
generate_ns_scanner_capability_proof(secret, challenge, server_epoch).expect("capability proof should be generated");
assert!(verify_ns_scanner_capability_proof(secret, challenge, server_epoch, &proof, false).is_ok());
assert!(verify_ns_scanner_capability_proof(secret, Uuid::new_v4(), server_epoch, &proof, false).is_err());
assert!(verify_ns_scanner_capability_proof(secret, challenge, Uuid::new_v4(), &proof, false).is_err());
assert!(verify_ns_scanner_capability_proof("different-secret", challenge, server_epoch, &proof, false).is_err());
}
#[test]
fn namespace_scanner_capability_proof_binds_tier_registry_generation_support() {
let secret = "test-scanner-capability-secret";
let challenge = Uuid::new_v4();
let server_epoch = Uuid::new_v4();
let proof = generate_ns_scanner_capability_proof(secret, challenge, server_epoch, true)
.expect("generation capability proof should be generated");
assert!(verify_ns_scanner_capability_proof(secret, challenge, server_epoch, &proof, true).is_ok());
assert!(verify_ns_scanner_capability_proof(secret, challenge, server_epoch, &proof, false).is_err());
let legacy = generate_ns_scanner_capability_proof(secret, challenge, server_epoch, false)
.expect("legacy capability proof should be generated");
assert!(verify_ns_scanner_capability_proof(secret, challenge, server_epoch, &legacy, true).is_err());
assert!(verify_ns_scanner_capability_proof(secret, challenge, server_epoch, &proof).is_ok());
assert!(verify_ns_scanner_capability_proof(secret, Uuid::new_v4(), server_epoch, &proof).is_err());
assert!(verify_ns_scanner_capability_proof(secret, challenge, Uuid::new_v4(), &proof).is_err());
assert!(verify_ns_scanner_capability_proof("different-secret", challenge, server_epoch, &proof).is_err());
}
/// Security regression for GHSA-r5qv-rc46-hv8q (internode RPC fail-closed,
@@ -13,18 +13,17 @@
// limitations under the License.
use crate::cluster::rpc::{
build_auth_headers, build_put_file_auth_trailer, verify_ns_scanner_capability_with_tier_registry_generation,
verify_put_file_capability,
build_auth_headers, build_put_file_auth_trailer, verify_ns_scanner_capability, verify_put_file_capability,
};
use crate::disk::error::{Error, Result};
use crate::disk::{FileReader, FileWriter};
use crate::storage_api_contracts::internode::{
NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY,
NS_SCANNER_PROTOCOL_VERSION, NS_SCANNER_PROTOCOL_VERSION_QUERY, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY,
NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY,
NsScannerCapabilityResponse, PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_CHALLENGE_QUERY,
PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION, PUT_FILE_NONCE_QUERY, PUT_FILE_SERVER_EPOCH_QUERY,
PutFileCapabilityResponse, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerCapabilityResponse, PUT_FILE_AUTH_QUERY,
PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION,
PUT_FILE_NONCE_QUERY, PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse, WALK_DIR_BODY_SHA256_QUERY,
WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
};
use async_trait::async_trait;
use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
@@ -138,12 +137,6 @@ fn put_file_capability_status_is_legacy(status: u16) -> bool {
status == 404
}
fn ns_scanner_capability_error_allows_legacy(error: &Error) -> bool {
[400, 404, 405, 426]
.into_iter()
.any(|status| error.is_internode_http_status(status))
}
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
#[allow(
dead_code,
@@ -227,7 +220,6 @@ pub struct NsScannerStreamRequest {
#[derive(Debug, Clone)]
pub struct NsScannerCapabilityRequest {
pub endpoint: String,
pub supports_tier_registry_generation: bool,
}
/// Data-plane stream opener used by `RemoteDisk`.
@@ -260,15 +252,6 @@ pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
async fn probe_ns_scanner(&self, _request: NsScannerCapabilityRequest) -> Result<Uuid> {
Err(Error::MethodNotAllowed)
}
async fn probe_ns_scanner_capability(&self, request: NsScannerCapabilityRequest) -> Result<NsScannerCapabilityResponse> {
let server_epoch = self.probe_ns_scanner(request).await?;
Ok(NsScannerCapabilityResponse {
version: NS_SCANNER_PROTOCOL_VERSION,
server_epoch,
proof: Vec::new(),
supports_tier_registry_generation: None,
})
}
// Interface facet nobody calls yet: every transport implements both, but no
// caller negotiates on them. Kept for the internode transport split
// (backlog#1350); deleting them would delete the seam and six impls.
@@ -352,44 +335,27 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
}
async fn probe_ns_scanner(&self, request: NsScannerCapabilityRequest) -> Result<Uuid> {
Ok(self.probe_ns_scanner_capability(request).await?.server_epoch)
}
async fn probe_ns_scanner_capability(&self, request: NsScannerCapabilityRequest) -> Result<NsScannerCapabilityResponse> {
if request.supports_tier_registry_generation {
return match self.probe_ns_scanner_capability_once(&request).await {
Ok(response) => Ok(response),
Err(marked_error) if ns_scanner_capability_error_allows_legacy(&marked_error) => {
// A v3 peer may reject the additive query marker, ignore
// it, or return its legacy proof. Retry once without the
// marker and only downgrade after that legacy response is
// authenticated; an unverified epoch is never trusted.
let legacy_request = NsScannerCapabilityRequest {
endpoint: request.endpoint.clone(),
supports_tier_registry_generation: false,
};
match self.probe_ns_scanner_capability_once(&legacy_request).await {
Ok(mut response) => {
response.supports_tier_registry_generation = None;
Ok(response)
}
Err(legacy_error) if ns_scanner_capability_error_allows_legacy(&legacy_error) => {
// Some old deployments expose only the legacy
// protocol response (or advertise 426). Treat
// the pair as an explicit unsupported result so
// the scanner can use its coordinator fallback.
Err(Error::MethodNotAllowed)
}
Err(_) => Err(marked_error),
}
}
// A server failure, network failure, or authentication error
// is not evidence of an old parser. Do not issue an
// unauthenticated legacy probe or silently downgrade.
Err(marked_error) => Err(marked_error),
};
let challenge = Uuid::new_v4();
let url = build_ns_scanner_capability_url(&request, challenge);
let mut headers = msgpack_headers();
build_auth_headers(&url, &Method::GET, &mut headers)?;
let reader = HttpReader::new(url, Method::GET, headers, None).await?;
let mut body = Vec::new();
reader
.take(u64::try_from(NS_SCANNER_MAX_CAPABILITY_RESPONSE_SIZE + 1).unwrap_or(u64::MAX))
.read_to_end(&mut body)
.await?;
if body.is_empty() || body.len() > NS_SCANNER_MAX_CAPABILITY_RESPONSE_SIZE {
return Err(Error::other("invalid remote namespace scanner capability response size"));
}
self.probe_ns_scanner_capability_once(&request).await
let response: NsScannerCapabilityResponse =
rmp_serde::from_slice(&body).map_err(|_| Error::other("invalid remote namespace scanner capability response"))?;
if response.version != NS_SCANNER_PROTOCOL_VERSION || response.server_epoch.is_nil() {
return Err(Error::other("incompatible remote namespace scanner capability response"));
}
verify_ns_scanner_capability(challenge, response.server_epoch, &response.proof)
.map_err(|err| Error::other(format!("remote namespace scanner capability authentication failed: {err}")))?;
Ok(response.server_epoch)
}
fn name(&self) -> &'static str {
@@ -402,53 +368,6 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
}
impl TcpHttpInternodeDataTransport {
async fn probe_ns_scanner_capability_once(
&self,
request: &NsScannerCapabilityRequest,
) -> Result<NsScannerCapabilityResponse> {
let challenge = Uuid::new_v4();
let url = build_ns_scanner_capability_url(request, challenge);
let mut headers = msgpack_headers();
build_auth_headers(&url, &Method::GET, &mut headers)?;
let reader = HttpReader::new(url, Method::GET, headers, None).await?;
let mut body = Vec::new();
reader
.take(u64::try_from(NS_SCANNER_MAX_CAPABILITY_RESPONSE_SIZE + 1).unwrap_or(u64::MAX))
.read_to_end(&mut body)
.await?;
if body.is_empty() || body.len() > NS_SCANNER_MAX_CAPABILITY_RESPONSE_SIZE {
return Err(Error::other("invalid remote namespace scanner capability response size"));
}
let mut response: NsScannerCapabilityResponse =
rmp_serde::from_slice(&body).map_err(|_| Error::other("invalid remote namespace scanner capability response"))?;
if response.version != NS_SCANNER_PROTOCOL_VERSION || response.server_epoch.is_nil() {
return Err(Error::other("incompatible remote namespace scanner capability response"));
}
if let Err(err) = verify_ns_scanner_capability_with_tier_registry_generation(
challenge,
response.server_epoch,
&response.proof,
request.supports_tier_registry_generation,
) {
// A permissive older peer can ignore the additive marker and
// return a valid legacy-scope proof with HTTP 200. Accept that
// response only after independently authenticating the legacy
// scope; all other verification failures remain fail-closed.
if request.supports_tier_registry_generation && ns_scanner_capability_legacy_proof_is_valid(challenge, &response) {
response.supports_tier_registry_generation = None;
return Ok(response);
}
return Err(Error::other(format!("remote namespace scanner capability authentication failed: {err}")));
}
// The proof authenticates the requested capability scope, not the
// optional response field. Derive the client-facing bit from that
// verified scope so an intermediary cannot strip or rewrite the field
// and force a silent downgrade after a successful generation-bound
// handshake.
normalize_ns_scanner_capability_response(&mut response, request.supports_tier_registry_generation);
Ok(response)
}
async fn put_file_auth_capability(&self, endpoint: &str) -> Result<Option<Uuid>> {
resolve_put_file_auth_capability(endpoint, || async {
tokio::time::timeout(PUT_FILE_CAPABILITY_PROBE_TIMEOUT, self.probe_put_file_auth(endpoint))
@@ -730,14 +649,6 @@ fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String {
)
}
fn normalize_ns_scanner_capability_response(response: &mut NsScannerCapabilityResponse, requested_generation_support: bool) {
response.supports_tier_registry_generation = requested_generation_support.then_some(true);
}
fn ns_scanner_capability_legacy_proof_is_valid(challenge: Uuid, response: &NsScannerCapabilityResponse) -> bool {
verify_ns_scanner_capability_with_tier_registry_generation(challenge, response.server_epoch, &response.proof, false).is_ok()
}
fn build_ns_scanner_url(request: &NsScannerStreamRequest) -> String {
let body_sha256 = hex_simd::encode_to_string(Sha256::digest(&request.body), hex_simd::AsciiCase::Lower);
format!(
@@ -764,18 +675,13 @@ fn build_ns_scanner_url(request: &NsScannerStreamRequest) -> String {
fn build_ns_scanner_capability_url(request: &NsScannerCapabilityRequest, challenge: Uuid) -> String {
format!(
"{}{}?{}={}&{}={}{}",
"{}{}?{}={}&{}={}",
request.endpoint,
NS_SCANNER_PATH,
NS_SCANNER_PROTOCOL_VERSION_QUERY,
NS_SCANNER_PROTOCOL_VERSION,
NS_SCANNER_CAPABILITY_CHALLENGE_QUERY,
challenge,
if request.supports_tier_registry_generation {
format!("&{}=true", NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY)
} else {
String::new()
}
challenge
)
}
@@ -888,7 +794,6 @@ mod tests {
let probe_err = transport
.probe_ns_scanner(NsScannerCapabilityRequest {
endpoint: "http://node1:9000".to_string(),
supports_tier_registry_generation: false,
})
.await
.expect_err("legacy transport should report namespace scanner as unsupported");
@@ -1482,7 +1387,6 @@ mod tests {
let url = build_ns_scanner_capability_url(
&NsScannerCapabilityRequest {
endpoint: "http://node1:9000".to_string(),
supports_tier_registry_generation: false,
},
challenge,
);
@@ -1495,85 +1399,6 @@ mod tests {
);
}
#[test]
fn ns_scanner_capability_url_marks_generation_support_only_when_requested() {
let challenge = Uuid::new_v4();
let url = build_ns_scanner_capability_url(
&NsScannerCapabilityRequest {
endpoint: "http://node1:9000".to_string(),
supports_tier_registry_generation: true,
},
challenge,
);
assert!(url.contains(&format!("&{}=true", NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY)));
}
#[test]
fn ns_scanner_capability_legacy_fallback_requires_explicit_compatibility_status() {
for status in [400, 404, 405, 426] {
let error = Error::from(rustfs_rio::new_test_internode_http_io_error(
rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::from_u16(status).expect("test status")),
));
assert!(
ns_scanner_capability_error_allows_legacy(&error),
"status {status} should permit legacy retry"
);
}
let marked_server_error = Error::from(rustfs_rio::new_test_internode_http_io_error(
rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::INTERNAL_SERVER_ERROR),
));
let network_error = Error::from(rustfs_rio::new_test_internode_http_io_error(
rustfs_rio::InternodeHttpErrorKind::ConnectionRefused,
));
let authentication_error = Error::other("remote namespace scanner capability authentication failed");
assert!(!ns_scanner_capability_error_allows_legacy(&marked_server_error));
assert!(!ns_scanner_capability_error_allows_legacy(&network_error));
assert!(!ns_scanner_capability_error_allows_legacy(&authentication_error));
}
#[test]
fn authenticated_ns_scanner_capability_ignores_unprotected_response_bit() {
let mut response = NsScannerCapabilityResponse {
version: NS_SCANNER_PROTOCOL_VERSION,
server_epoch: Uuid::new_v4(),
proof: Vec::new(),
supports_tier_registry_generation: None,
};
normalize_ns_scanner_capability_response(&mut response, true);
assert_eq!(response.supports_tier_registry_generation, Some(true));
response.supports_tier_registry_generation = Some(false);
normalize_ns_scanner_capability_response(&mut response, false);
assert_eq!(response.supports_tier_registry_generation, None);
}
#[test]
fn ns_scanner_capability_accepts_only_authenticated_legacy_scope_after_marker_mismatch() {
crate::runtime::sources::ensure_test_rpc_secret();
let challenge = Uuid::new_v4();
let response = NsScannerCapabilityResponse {
version: NS_SCANNER_PROTOCOL_VERSION,
server_epoch: Uuid::new_v4(),
proof: crate::cluster::rpc::sign_ns_scanner_capability(challenge, Uuid::new_v4())
.expect("placeholder proof should be generated"),
supports_tier_registry_generation: None,
};
// A proof bound to a different challenge cannot authorize the legacy
// fallback, even though the response has the expected shape.
assert!(!ns_scanner_capability_legacy_proof_is_valid(challenge, &response));
let server_epoch = response.server_epoch;
let valid_response = NsScannerCapabilityResponse {
proof: crate::cluster::rpc::sign_ns_scanner_capability(challenge, server_epoch)
.expect("legacy proof should be generated"),
..response
};
assert!(ns_scanner_capability_legacy_proof_is_valid(challenge, &valid_response));
}
#[test]
fn transport_config_defaults_to_tcp_http() {
let transport = build_internode_data_transport(None).unwrap();
+2 -3
View File
@@ -35,9 +35,8 @@ pub use http_auth::{
TONIC_RPC_PREFIX, build_auth_headers, build_put_file_auth_trailer, check_and_record_signed_rpc_nonce, gen_signature_headers,
gen_tonic_replay_scope_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience, set_tonic_canonical_body_digest,
set_tonic_mutation_body_digest, set_tonic_rolling_canonical_body_digest, set_tonic_rolling_mutation_body_digest,
sign_ns_scanner_capability, sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability,
sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_ns_scanner_capability, verify_ns_scanner_capability_with_tier_registry_generation, verify_put_file_auth_trailer,
sign_ns_scanner_capability, sign_put_file_capability, sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge,
tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason, verify_ns_scanner_capability, verify_put_file_auth_trailer,
verify_put_file_capability, verify_rpc_signature, verify_tonic_boot_epoch_response, verify_tonic_canonical_body_digest,
verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
verify_tonic_rpc_signature_with_bootstrap,
+3 -58
View File
@@ -781,16 +781,14 @@ impl RemoteDisk {
if self.health.is_faulty() {
return Err(DiskError::FaultyDisk);
}
let probe = self.data_transport.probe_ns_scanner_capability(NsScannerCapabilityRequest {
let probe = self.data_transport.probe_ns_scanner(NsScannerCapabilityRequest {
endpoint: self.endpoint.grid_host(),
supports_tier_registry_generation: true,
});
let result = timeout(NS_SCANNER_CAPABILITY_PROBE_TIMEOUT, probe)
.await
.map_err(|_| DiskError::other("remote namespace scanner capability probe timed out"))?;
match result {
Ok(response) if response.supports_tier_registry_generation == Some(true) => Ok(Some(response.server_epoch)),
Ok(_) => Ok(None),
Ok(server_epoch) => Ok(Some(server_epoch)),
// RUSTFS_COMPAT_TODO(ns-scanner-rpc-v3): old peers and legacy transports lack the authenticated startup-epoch handshake. Remove after every supported peer implements namespace scanner protocol v3.
Err(DiskError::MethodNotAllowed) => Ok(None),
Err(err)
@@ -4042,21 +4040,10 @@ mod tests {
NsScannerProbe(NsScannerCapabilityRequest),
}
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Default)]
struct RecordingInternodeDataTransport {
calls: Arc<StdMutex<Vec<RecordedTransportCall>>>,
ns_scanner_probe_status: Arc<StdMutex<Option<u16>>>,
ns_scanner_generation_support: Arc<StdMutex<Option<bool>>>,
}
impl Default for RecordingInternodeDataTransport {
fn default() -> Self {
Self {
calls: Arc::default(),
ns_scanner_probe_status: Arc::default(),
ns_scanner_generation_support: Arc::new(StdMutex::new(Some(true))),
}
}
}
#[derive(Clone, Debug)]
@@ -4276,15 +4263,6 @@ mod tests {
Self {
calls: Arc::default(),
ns_scanner_probe_status: Arc::new(StdMutex::new(Some(status))),
ns_scanner_generation_support: Arc::new(StdMutex::new(Some(true))),
}
}
fn with_ns_scanner_generation_support(support: Option<bool>) -> Self {
Self {
calls: Arc::default(),
ns_scanner_probe_status: Arc::default(),
ns_scanner_generation_support: Arc::new(StdMutex::new(support)),
}
}
@@ -4967,23 +4945,6 @@ mod tests {
Ok(Uuid::from_u128(1))
}
async fn probe_ns_scanner_capability(
&self,
request: NsScannerCapabilityRequest,
) -> Result<crate::storage_api_contracts::internode::NsScannerCapabilityResponse> {
let server_epoch = self.probe_ns_scanner(request).await?;
let supports_tier_registry_generation = *self
.ns_scanner_generation_support
.lock()
.expect("namespace scanner generation support lock poisoned");
Ok(crate::storage_api_contracts::internode::NsScannerCapabilityResponse {
version: crate::storage_api_contracts::internode::NS_SCANNER_PROTOCOL_VERSION,
server_epoch,
proof: Vec::new(),
supports_tier_registry_generation,
})
}
fn name(&self) -> &'static str {
"recording"
}
@@ -6796,22 +6757,6 @@ mod tests {
}
}
#[tokio::test]
async fn test_remote_disk_namespace_scanner_capability_falls_back_without_generation_support() {
for support in [None, Some(false)] {
let transport = RecordingInternodeDataTransport::with_ns_scanner_generation_support(support);
let remote_disk = new_remote_disk_with_transport(Arc::new(transport)).await;
assert_eq!(
remote_disk
.ns_scanner_server_epoch()
.await
.expect("missing generation support should be classified as unsupported"),
None
);
}
}
#[tokio::test]
async fn test_remote_disk_namespace_scanner_capability_rejects_legacy_transport() {
let remote_disk = new_remote_disk_with_transport(Arc::new(RetryingOpenReadInternodeDataTransport::default())).await;
+6 -38
View File
@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
///
/// # Returns
/// A Result containing the BitrotWriterWrapper or an error
/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
///
/// A known length is grown by one checksum per shard so the on-disk file size
/// matches what the bitrot writer emits. A negative length is the
/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
/// a fixed body length when locating the authenticated trailer. Clamping it
/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
/// because a genuinely empty object still means an empty body.
fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
if length <= 0 {
return length;
}
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
}
pub async fn create_bitrot_writer(
is_inline_buffer: bool,
disk: Option<&DiskStore>,
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
let writer = if is_inline_buffer {
CustomWriter::new_inline_buffer()
} else if let Some(disk) = disk {
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
let length = if length > 0 {
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
} else {
0
};
let file = disk.create_file("", volume, path, length).await?;
#[cfg(feature = "hotpath")]
@@ -833,25 +820,6 @@ mod tests {
use rustfs_rio::ChunkReader;
use std::collections::VecDeque;
#[test]
fn bitrot_create_file_size_grows_known_length_by_checksums() {
// 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
}
#[test]
fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
// SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
// put_file_stream receiver relies on a non-positive size to parse the auth
// trailer from the stream tail, so the sentinel must survive untouched.
assert_eq!(
bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
);
}
struct TestChunkReader {
chunks: VecDeque<Bytes>,
}
+14 -1
View File
@@ -2124,13 +2124,26 @@ impl SetDisks {
let put_object_size = known_put_object_storage_size(data.size());
let shard_file_size_raw = erasure.shard_file_size(put_object_size);
let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let is_inline_buffer =
storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
let shard_file_size = shard_file_size_raw;
let shard_size = erasure.shard_size();
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
{
use std::io::Write;
let msg = format!(
"INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
);
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
let _ = f.write_all(msg.as_bytes());
}
let _ = std::io::stderr().write_all(msg.as_bytes());
}
rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
let (mut writers, errors) = if direct_inline_commit {
@@ -27,12 +27,12 @@ pub(crate) mod internode {
NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY,
NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_PROTOCOL_VERSION, NS_SCANNER_PROTOCOL_VERSION_QUERY,
NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY,
NS_SCANNER_SESSION_SEQUENCE_QUERY, NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY, NsScannerCapabilityResponse,
PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_TRAILER_DIGEST_LEN, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_TRAILER_MAC_LEN,
PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY,
PUT_FILE_CAPABILITY_VERSION, PUT_FILE_NONCE_QUERY, PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse,
SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerCapabilityResponse, PUT_FILE_AUTH_QUERY, PUT_FILE_AUTH_TRAILER_DIGEST_LEN,
PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_TRAILER_MAC_LEN, PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_AUTH_V1,
PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, PUT_FILE_CAPABILITY_VERSION, PUT_FILE_NONCE_QUERY,
PUT_FILE_SERVER_EPOCH_QUERY, PutFileCapabilityResponse, SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION,
SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION, WALK_DIR_BODY_SHA256_QUERY,
WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
};
}
+1 -1
View File
@@ -3194,7 +3194,7 @@ impl ECStore {
// Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let accounting = vec![None; objects.len()];
let mut accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
.filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
.collect::<Vec<_>>();
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+25 -86
View File
@@ -16,14 +16,14 @@
//!
//! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
//! kills the active node while this test continuously decrypts through a
//! surviving standby. KV2 and Transit must recover after the bounded circuit
//! interval, use a bounded number of attempts, and leave the circuit and
//! in-flight gauges at zero after a new leader is elected.
//! surviving standby. KV2 and Transit requests must remain successful, use a
//! bounded number of attempts, and leave the circuit and in-flight gauges at
//! zero after a new leader is elected.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use metrics_util::MetricKind;
@@ -43,11 +43,6 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
const MAX_ATTEMPTS: u32 = 10;
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
// The circuit remains open for 30s after five failed attempts.
const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
type MetricEntry = (
metrics_util::CompositeKey,
@@ -69,7 +64,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
backend,
backend_config,
allow_insecure_dev_defaults: true,
timeout: ATTEMPT_TIMEOUT,
timeout: Duration::from_secs(2),
retry_attempts: MAX_ATTEMPTS,
enable_cache: false,
..KmsConfig::default()
@@ -169,31 +164,14 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
.sum()
}
async fn wait_for_count(
counter: &AtomicU64,
failure: &Mutex<Option<String>>,
minimum: u64,
description: &str,
timeout: Duration,
) {
tokio::time::timeout(timeout, async {
async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
tokio::time::timeout(Duration::from_secs(20), async {
while counter.load(Ordering::SeqCst) < minimum {
if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
panic!(
"{description} worker failed after {} successful decrypts: {error}",
counter.load(Ordering::SeqCst)
);
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap_or_else(|_| {
panic!(
"timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
counter.load(Ordering::SeqCst)
)
});
.unwrap_or_else(|_| panic!("timed out waiting for {description}"));
}
async fn wait_for_file(path: &Path, description: &str) {
@@ -211,8 +189,7 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
request: DecryptRequest,
expected: Vec<u8>,
completed: Arc<AtomicU64>,
allow_failover_errors: Arc<AtomicBool>,
failure: Arc<Mutex<Option<String>>>,
failed: Arc<AtomicBool>,
stop: CancellationToken,
) {
while !stop.is_cancelled() {
@@ -220,18 +197,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
Ok(response) if response.plaintext == expected => {
completed.fetch_add(1, Ordering::SeqCst);
}
Ok(_) => {
*failure.lock().expect("decrypt failure lock poisoned") =
Some("decrypt returned unexpected plaintext".to_string());
return;
}
Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
if allow_failover_errors.load(Ordering::SeqCst) =>
{
tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
}
Err(error) => {
*failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
Ok(_) | Err(_) => {
failed.store(true, Ordering::SeqCst);
return;
}
}
@@ -329,9 +296,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
);
let stop = CancellationToken::new();
let allow_failover_errors = Arc::new(AtomicBool::new(false));
let kv2_failure = Arc::new(Mutex::new(None));
let transit_failure = Arc::new(Mutex::new(None));
let failed = Arc::new(AtomicBool::new(false));
let kv2_completed = Arc::new(AtomicU64::new(0));
let transit_completed = Arc::new(AtomicU64::new(0));
let kv2_worker = tokio::spawn(decrypt_loop(
@@ -339,8 +304,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
kv2_request,
kv2_data_key.plaintext_key,
Arc::clone(&kv2_completed),
Arc::clone(&allow_failover_errors),
Arc::clone(&kv2_failure),
Arc::clone(&failed),
stop.clone(),
));
let transit_worker = tokio::spawn(decrypt_loop(
@@ -348,21 +312,12 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
transit_request,
transit_data_key.plaintext_key,
Arc::clone(&transit_completed),
Arc::clone(&allow_failover_errors),
Arc::clone(&transit_failure),
Arc::clone(&failed),
stop.clone(),
));
wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
wait_for_count(
&transit_completed,
&transit_failure,
2,
"two healthy Transit decrypts",
HEALTHY_PROGRESS_TIMEOUT,
)
.await;
allow_failover_errors.store(true, Ordering::SeqCst);
wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
wait_for_file(&elected, "the replacement Vault leader").await;
@@ -371,39 +326,18 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
wait_for_count(
&kv2_completed,
&kv2_failure,
kv2_after_election,
"post-failover KV2 decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
wait_for_count(
&transit_completed,
&transit_failure,
transit_after_election,
"post-failover Transit decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
stop.cancel();
kv2_worker.await.expect("KV2 decrypt worker must join");
transit_worker.await.expect("Transit decrypt worker must join");
assert!(
kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
"no KV2 decrypt may fail or return different plaintext"
);
assert!(
transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
"no Transit decrypt may fail or return different plaintext"
);
assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
}
#[test]
#[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
@@ -415,6 +349,11 @@ fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
});
let snapshot = snapshotter.snapshot().into_vec();
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
0,
"a bounded leader election must not open the circuit"
);
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
0,
+12 -133
View File
@@ -29,8 +29,7 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
pub use rustfs_data_usage::{
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierAccountingProof, TierStats, UNKNOWN_TIER,
UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP, UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP, UnknownTierStats, hash_path, prefix_usage_in_cache,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
@@ -206,79 +205,25 @@ impl ScannerSizeSummaryExt for SizeSummary {
self.versions = self.versions.saturating_add(1);
}
let logical_size = size.max(0);
let size = usize::try_from(logical_size).unwrap_or(usize::MAX);
let size = usize::try_from(size.max(0)).unwrap_or(usize::MAX);
self.total_size = self.total_size.saturating_add(size);
let logical_bytes = u64::try_from(logical_size).unwrap_or(u64::MAX);
let physical_bytes = u64::try_from(oi.size.max(0)).unwrap_or(0);
let mut proof = TierAccountingProof {
logical_total: logical_bytes,
logical_known: 0,
physical_total: physical_bytes,
physical_known: 0,
overflowed: false,
};
if oi.transitioned_object.free_version {
proof.logical_known = logical_bytes;
proof.physical_known = physical_bytes;
self.tier_accounting_proof.saturating_add(proof);
return;
}
let tier = if oi.transitioned_object.status == TRANSITION_COMPLETE {
oi.transitioned_object.tier.as_str()
} else {
oi.storage_class.as_deref().unwrap_or(storageclass::STANDARD)
};
let builtin_tier = tier == storageclass::STANDARD || tier == storageclass::RRS;
let tier_registry_is_empty =
self.tier_stats.is_empty() || (self.tier_stats.len() == 1 && self.tier_stats.contains_key(UNKNOWN_TIER));
let known_tier = tier != UNKNOWN_TIER && (builtin_tier || self.tier_stats.contains_key(tier));
// With no configured tier, retain the historical empty-map shape for
// ordinary STANDARD/RRS objects. A non-built-in key is still an
// observable unknown and must create only the fixed bucket.
if tier_registry_is_empty && known_tier {
proof.logical_known = logical_bytes;
proof.physical_known = physical_bytes;
self.tier_accounting_proof.saturating_add(proof);
return;
let mut tier = oi.storage_class.clone().unwrap_or_else(|| storageclass::STANDARD.to_string());
if oi.transitioned_object.status == TRANSITION_COMPLETE {
tier = oi.transitioned_object.tier.clone();
}
// Configured tiers and the fixed bucket are normally seeded, so the
// hot path can mutate them without allocating a key for every object.
// The fallback inserts only when a legacy/no-config summary sees its
// first unknown key.
let tier_stats = if known_tier {
if let Some(stats) = self.tier_stats.get_mut(tier) {
stats
} else {
self.tier_stats.entry(tier.to_owned()).or_default()
}
} else if let Some(stats) = self.tier_stats.get_mut(UNKNOWN_TIER) {
stats
} else {
self.tier_stats.entry(UNKNOWN_TIER.to_string()).or_default()
};
*tier_stats = tier_stats.add(&TierStats {
total_size: physical_bytes,
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
if known_tier {
proof.logical_known = logical_bytes;
proof.physical_known = physical_bytes;
if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
*tier_stats = tier_stats.add(&TierStats {
total_size: u64::try_from(oi.size).unwrap_or(0),
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
}
if !known_tier {
self.unknown_tier_stats
.record_dimensions(tier, logical_bytes, physical_bytes, 1, u64::from(oi.is_latest));
if self.unknown_tier_stats.counter_overflowed {
proof.overflowed = true;
}
}
self.tier_accounting_proof.saturating_add(proof);
}
}
@@ -326,11 +271,6 @@ pub struct DataUsageEntryInfo {
pub name: String,
pub parent: String,
pub entry: DataUsageEntry,
/// Registry generation used to classify this root entry. Older remote
/// workers omit it; callers must reject that result when a frozen cycle
/// requires generation fencing.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tier_registry_generation: Option<u64>,
}
#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq, Hash)]
@@ -404,10 +344,6 @@ pub struct DataUsageCacheInfo {
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
#[serde(default)]
pub cache_key_format: u16,
/// Registry generation used for the completed/partial scan. This is
/// process-local audit data; older cache writers omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tier_registry_generation: Option<u64>,
}
impl Serialize for DataUsageCacheInfo {
@@ -417,8 +353,7 @@ impl Serialize for DataUsageCacheInfo {
{
// Keep this metadata map-encoded so older readers can ignore fields
// appended by newer scanner versions during rolling upgrades.
let field_count = 16 + usize::from(self.tier_registry_generation.is_some());
let mut state = serializer.serialize_map(Some(field_count))?;
let mut state = serializer.serialize_map(Some(16))?;
state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?;
state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -435,9 +370,6 @@ impl Serialize for DataUsageCacheInfo {
state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
state.serialize_entry("cache_key_format", &self.cache_key_format)?;
if let Some(generation) = self.tier_registry_generation {
state.serialize_entry("tier_registry_generation", &generation)?;
}
state.end()
}
}
@@ -458,58 +390,6 @@ pub(crate) enum DataUsageCachePrepareOutcome {
}
impl DataUsageCache {
/// Reconcile tier keys loaded from an older cache against the registry
/// frozen for this scan. New metadata is already routed through
/// `UNKNOWN_TIER`; this pass handles retired keys that predate that rule.
/// Legacy `TierStats` carries physical bytes only, so this migration does
/// not manufacture a logical unknown-byte value from that physical total.
pub(crate) fn fold_retired_tiers(&mut self, tier_names: &[String]) {
let known_tiers = tier_names.iter().map(String::as_str).collect::<HashSet<_>>();
for entry in self.cache.values_mut() {
let Some(tiers) = entry.all_tier_stats.as_mut() else { continue };
let existing_unknown = tiers.tiers.get(UNKNOWN_TIER).cloned().unwrap_or_default();
let companion_present = entry.unknown_tier_stats.as_ref().is_some_and(|stats| !stats.is_empty());
let migrate_existing_unknown = !companion_present;
let mut retired = TierStats::default();
let mut retired_key_found = false;
if migrate_existing_unknown {
retired = retired.add(&existing_unknown);
}
for (tier, stats) in &tiers.tiers {
if tier != UNKNOWN_TIER
&& tier != storageclass::STANDARD
&& tier != storageclass::RRS
&& !known_tiers.contains(tier.as_str())
{
retired_key_found = true;
retired = retired.add(stats);
}
}
tiers.fold_unknown_tiers(tier_names.iter().map(String::as_str));
if !retired.is_empty() && !companion_present {
entry.add_unknown_tier_stats(&UnknownTierStats {
// The legacy map stores physical bytes only. Logical
// bytes remain zero until a fresh object scan observes
// them under the current metadata format.
unknown_physical_bytes: retired.total_size,
unknown_objects: retired.num_objects,
unknown_versions: retired.num_versions,
..Default::default()
});
// The legacy tier map has no logical-byte dimension, so a
// proof that classified this retired key as known cannot be
// repaired safely. Mark it unvalidated and require a fresh
// scan rather than guessing a logical subtraction.
entry.tier_accounting_proof = None;
} else if retired_key_found {
// A nonempty companion has no provenance tying it to the
// retired map keys. Reject the mixed cache until a fresh scan
// reconciles the dimensions instead of double-counting them.
entry.tier_accounting_proof = None;
}
}
}
/// Prefix-level usage query over this (writer-side) cache; see
/// [`prefix_usage_in_cache`] for the semantics
/// (rustfs/backlog#1872).
@@ -995,7 +875,6 @@ impl DataUsageCache {
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: flat.unknown_tier_stats.filter(|stats| !stats.is_empty()),
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
buckets_usage,
..Default::default()
+1 -263
View File
@@ -573,6 +573,7 @@ fn size_summary_add_saturates_all_usage_counters() {
failed_count: usize::MAX,
},
);
let mut increment = SizeSummary {
total_size: 1,
versions: 1,
@@ -587,24 +588,6 @@ fn size_summary_add_saturates_all_usage_counters() {
failed_count: 1,
..Default::default()
};
summary.tier_stats.insert(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: u64::MAX,
num_versions: u64::MAX,
num_objects: u64::MAX,
},
);
increment.tier_stats.insert(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: 1,
num_versions: 1,
num_objects: 1,
},
);
increment.unknown_tier_stats.unknown_bytes = 1;
increment.unknown_tier_stats.unknown_physical_bytes = 1;
increment.repl_target_stats.insert(
target.clone(),
ReplTargetSizeSummary {
@@ -641,8 +624,6 @@ fn size_summary_add_saturates_all_usage_counters() {
assert_eq!(target_summary.failed_size, i64::MAX);
assert_eq!(target_summary.pending_count, usize::MAX);
assert_eq!(target_summary.failed_count, usize::MAX);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, u64::MAX);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1);
}
#[test]
@@ -692,249 +673,6 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
);
}
#[test]
fn unknown_tier_is_bounded_and_accounted() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
size: 11,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&object, 11, 11);
assert_eq!(summary.tier_stats.len(), 2);
assert_eq!(summary.tier_stats.get(UNKNOWN_TIER).map(|stats| stats.total_size), Some(11));
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 11);
assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 11);
assert_eq!(summary.unknown_tier_stats.unknown_objects, 1);
assert_eq!(summary.tier_accounting_proof.logical_total, 11);
assert_eq!(summary.tier_accounting_proof.logical_known, 0);
assert_eq!(summary.tier_accounting_proof.physical_total, 11);
assert_eq!(summary.tier_accounting_proof.physical_known, 0);
assert!(summary.unknown_tier_stats.diagnostics.len() <= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP);
assert!(summary.unknown_tier_stats.diagnostics.iter().map(String::len).sum::<usize>() <= UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP);
assert!(
summary
.unknown_tier_stats
.diagnostics
.iter()
.all(|entry| !entry.contains("retired"))
);
}
#[test]
fn unknown_tier_is_accounted_when_no_remote_tier_is_configured() {
let mut summary = SizeSummary::new();
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
size: 3,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&object, 9, 9);
assert_eq!(summary.tier_stats.len(), 1);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 3);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 9);
assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 3);
let standard = ObjectInfo {
storage_class: Some(storageclass::STANDARD.to_string()),
size: 4,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&standard, 4, 4);
assert_eq!(summary.tier_accounting_proof.logical_total, 13);
assert_eq!(summary.tier_accounting_proof.logical_known, 4);
assert_eq!(summary.tier_accounting_proof.physical_total, 3 + 4);
assert_eq!(summary.tier_accounting_proof.physical_known, 4);
assert_eq!(summary.tier_stats.len(), 1, "built-ins preserve the no-tier map shape");
}
#[test]
fn million_unique_tier_keys_do_not_grow_stats_map() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
for index in 0..1_000_000_u64 {
let object = ObjectInfo {
storage_class: Some(format!("untrusted-tier-{index}")),
size: 1,
..Default::default()
};
summary.actions_accounting(&object, 1, 1);
}
assert_eq!(summary.tier_stats.len(), 2);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 1_000_000);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1_000_000);
assert!(summary.unknown_tier_stats.diagnostics.len() <= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP);
assert!(summary.unknown_tier_stats.diagnostics.iter().map(String::len).sum::<usize>() <= UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP);
}
#[test]
fn unknown_tier_never_triggers_transition() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
let mut object = ObjectInfo {
storage_class: Some("removed-tier".to_string()),
size: 7,
..Default::default()
};
object.transitioned_object.status = TRANSITION_COMPLETE.to_string();
object.transitioned_object.tier = "removed-tier".to_string();
summary.actions_accounting(&object, 7, 7);
assert_eq!(summary.tier_stats.get("removed-tier"), None);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 7);
}
#[test]
fn removed_tier_survives_restart_as_unknown() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("COLD".to_string(), TierStats::default());
summary.tier_stats.insert(
"RETIRED".to_string(),
TierStats {
total_size: 5,
num_versions: 1,
num_objects: 1,
},
);
let object = ObjectInfo {
storage_class: Some("COLD".to_string()),
size: 5,
..Default::default()
};
summary.actions_accounting(&object, 5, 5);
let mut entry = DataUsageEntry::default();
entry.add_tier_sizes(&summary.tier_stats);
entry.add_unknown_tier_stats(&UnknownTierStats {
unknown_bytes: 2,
unknown_physical_bytes: 2,
unknown_objects: 1,
unknown_versions: 1,
..Default::default()
});
let encoded = rmp_serde::to_vec(&entry).expect("entry should encode");
let restored: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("entry should decode");
assert_eq!(restored.unknown_tier_stats.as_ref().map(|stats| stats.unknown_bytes), Some(2));
assert_eq!(
restored.all_tier_stats.as_ref().expect("tier stats persisted").tiers["RETIRED"].total_size,
5
);
let mut cache = DataUsageCache::default();
cache.replace("bucket", "", restored);
cache.fold_retired_tiers(&["COLD".to_string()]);
let folded = cache.cache.get(&hash_path("bucket").key()).expect("folded cache entry");
assert_eq!(
folded.all_tier_stats.as_ref().expect("tier stats persisted").tiers[UNKNOWN_TIER].total_size,
5
);
assert_eq!(folded.unknown_tier_stats.as_ref().map(|stats| stats.unknown_bytes), Some(2));
}
#[test]
fn retired_tier_fold_is_idempotent_and_rejects_mixed_companion_provenance() {
let mut cache = DataUsageCache::default();
let mut entry = DataUsageEntry {
all_tier_stats: Some(AllTierStats {
tiers: HashMap::from([(
"RETIRED".to_string(),
TierStats {
total_size: 5,
num_versions: 1,
num_objects: 1,
},
)]),
}),
..Default::default()
};
entry.unknown_tier_stats = Some(UnknownTierStats {
unknown_physical_bytes: 5,
..Default::default()
});
entry.tier_accounting_proof = Some(TierAccountingProof {
physical_total: 5,
physical_known: 5,
..Default::default()
});
cache.replace("bucket", "", entry);
cache.fold_retired_tiers(&["COLD".to_string()]);
let first = cache.cache.get(&hash_path("bucket").key()).expect("entry").clone();
assert_eq!(first.all_tier_stats.as_ref().expect("tiers").tiers[UNKNOWN_TIER].total_size, 5);
assert_eq!(first.unknown_tier_stats.as_ref().expect("companion").unknown_physical_bytes, 5);
assert!(first.tier_accounting_proof.is_none(), "mixed provenance must not publish");
cache.fold_retired_tiers(&["COLD".to_string()]);
let second = cache.cache.get(&hash_path("bucket").key()).expect("entry");
assert_eq!(second.all_tier_stats.as_ref().expect("tiers").tiers[UNKNOWN_TIER].total_size, 5);
assert_eq!(second.unknown_tier_stats.as_ref().expect("companion").unknown_physical_bytes, 5);
}
#[test]
fn tier_registry_refresh_does_not_mix_cycle_generations() {
let first = crate::TierRegistrySnapshot {
generation: 1,
names: Arc::from(["WARM".to_string()]),
refresh_failed: false,
};
let second = crate::TierRegistrySnapshot {
generation: 2,
names: Arc::from(["COLD".to_string()]),
refresh_failed: false,
};
assert_ne!(first.generation, second.generation);
assert_eq!(first.names.as_ref(), ["WARM".to_string()]);
assert_eq!(second.names.as_ref(), ["COLD".to_string()]);
assert!(first.refreshed(Err(())).refresh_failed);
assert!(!second.refreshed(Ok(Arc::from(["HOT".to_string()]))).refresh_failed);
assert_eq!(first.refreshed(Err(())).generation, first.generation);
assert_eq!(first.refreshed(Err(())).names, first.names);
}
#[test]
fn unknown_tier_counter_uses_checked_arithmetic() {
let max = TierStats {
total_size: u64::MAX,
num_versions: u64::MAX,
num_objects: u64::MAX,
};
assert!(max.checked_add(&TierStats::default()).is_some());
assert!(
max.checked_add(&TierStats {
total_size: 1,
..Default::default()
})
.is_none()
);
let mut unknown = UnknownTierStats {
unknown_bytes: u64::MAX,
..Default::default()
};
unknown.record("overflow", 1, 1, 1);
assert_eq!(unknown.unknown_bytes, u64::MAX);
assert_eq!(unknown.unknown_objects, 1);
assert!(unknown.counter_overflowed);
assert!(unknown.checked_add(&UnknownTierStats::default()).is_none());
assert!(
unknown
.checked_add(&UnknownTierStats {
unknown_bytes: 1,
..Default::default()
})
.is_none()
);
}
#[test]
fn test_data_usage_entry_merge_sums_failed_objects() {
let mut left = DataUsageEntry {
+10 -286
View File
@@ -24,11 +24,8 @@
use bytes::Bytes;
use http::HeaderMap;
use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config};
use sha2::{Digest as _, Sha256};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::LazyLock;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use storage_api::owner::{
@@ -97,45 +94,6 @@ static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);
static SCANNER_FOREGROUND_READ_ACTIVITY: AtomicU64 = AtomicU64::new(0);
static SCANNER_FOREGROUND_STREAM_READS: AtomicU64 = AtomicU64::new(0);
/// Immutable tier registry captured at the beginning of a folder scan.
/// Generation makes it possible to prove that a result was classified against
/// one registry even when the process-wide TTL cache refreshes concurrently.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct TierRegistrySnapshot {
pub(crate) generation: u64,
pub(crate) names: Arc<[String]>,
/// True when the last refresh attempt failed and `names` is therefore a
/// retained last-good snapshot rather than a newly read registry.
pub(crate) refresh_failed: bool,
}
impl TierRegistrySnapshot {
/// Apply a refresh only when the registry read succeeds. A failed refresh
/// retains the prior generation, preventing a transient config failure
/// from classifying the remainder of a scan against an empty registry.
pub(crate) fn refreshed(&self, names: Result<Arc<[String]>, ()>) -> Self {
match names {
Ok(names) => Self {
generation: self.generation.saturating_add(1),
names,
refresh_failed: false,
},
Err(()) => Self {
refresh_failed: true,
..self.clone()
},
}
}
pub(crate) fn initial(names: Arc<[String]>) -> Self {
Self {
generation: 1,
names,
refresh_failed: false,
}
}
}
pub fn current_scanner_activity() -> u64 {
SCANNER_ACTIVE_WORK_UNITS.load(Ordering::Relaxed)
}
@@ -413,7 +371,6 @@ pub(crate) fn resolve_scanner_server_config() -> Option<ServerConfig> {
/// How long the scanner caches the runtime tier-name list before re-reading
/// the tier configuration manager.
const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30);
const MAX_TIER_REGISTRY_NAME_BYTES: usize = 256;
/// Process-wide TTL cache of runtime tier names.
///
@@ -426,192 +383,24 @@ const MAX_TIER_REGISTRY_NAME_BYTES: usize = 256;
/// `TIER_NAME_CACHE_TTL` later; a removed tier can leave an all-zero
/// `TierStats` seed behind for one cache generation, which merges harmlessly
/// by key in per-object accounting and disappears on the next refresh.
static TIER_NAME_CACHE: RwLock<Option<(Instant, TierRegistrySnapshot)>> = RwLock::new(None);
static TIER_REGISTRY_GENERATION: AtomicU64 = AtomicU64::new(0);
static TIER_CYCLE_SNAPSHOTS: LazyLock<RwLock<HashMap<(u64, u64), TierRegistrySnapshot>>> =
LazyLock::new(|| RwLock::new(HashMap::new()));
static TIER_ACTIVE_CYCLES: LazyLock<RwLock<HashMap<(u64, u64), usize>>> = LazyLock::new(|| RwLock::new(HashMap::new()));
static TIER_NAME_REFRESH_LOCK: LazyLock<tokio::sync::Mutex<()>> = LazyLock::new(|| tokio::sync::Mutex::new(()));
/// Return one immutable registry snapshot for a scanner unit of work.
pub(crate) async fn runtime_tier_registry() -> TierRegistrySnapshot {
{
let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
if let Some((refreshed_at, snapshot)) = cached
&& refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
{
return snapshot;
}
}
// Serialize refreshes so a slower read of the old config cannot overwrite
// a newer snapshot published by a concurrent caller.
let _refresh_guard = TIER_NAME_REFRESH_LOCK.lock().await;
{
let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
if let Some((refreshed_at, snapshot)) = cached
&& refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
{
return snapshot;
}
}
let previous = TIER_NAME_CACHE
.read()
.unwrap_or_else(|err| err.into_inner())
.as_ref()
.map(|(_, snapshot)| snapshot.clone());
let names = ecstore_get_global_tier_config_mgr()
.read()
.await
.list_tiers()
.into_iter()
.map(|tier| tier.name)
.collect::<Vec<_>>();
let snapshot = match validate_tier_registry_names(names) {
Ok(names) => {
let generation = next_tier_registry_generation();
match previous {
Some(previous) => TierRegistrySnapshot {
generation,
..previous.refreshed(Ok(names))
},
None => TierRegistrySnapshot {
generation,
..TierRegistrySnapshot::initial(names)
},
}
}
Err(()) => match previous {
Some(previous) => previous.refreshed(Err(())),
None => TierRegistrySnapshot {
generation: next_tier_registry_generation(),
names: Arc::new([]),
refresh_failed: true,
},
},
};
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), snapshot.clone()));
snapshot
}
fn next_tier_registry_generation() -> u64 {
TIER_REGISTRY_GENERATION
.fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| Some(current.saturating_add(1)))
.unwrap_or(u64::MAX)
}
fn validate_tier_registry_names(mut names: Vec<String>) -> Result<Arc<[String]>, ()> {
if names.iter().any(|name| {
name.is_empty()
|| name.len() > MAX_TIER_REGISTRY_NAME_BYTES
|| name.bytes().any(|byte| byte.is_ascii_control())
|| name == UNKNOWN_TIER
|| name == storageclass::STANDARD
|| name == storageclass::RRS
}) {
return Err(());
}
names.sort_unstable();
if names.windows(2).any(|pair| pair[0] == pair[1]) {
return Err(());
}
Ok(names.into())
}
static TIER_NAME_CACHE: RwLock<Option<(Instant, Arc<[String]>)>> = RwLock::new(None);
/// Tier names currently registered in the tier configuration, cached for
/// `TIER_NAME_CACHE_TTL`.
pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
runtime_tier_registry().await.names
}
/// Return the immutable tier registry for one scanner cycle/leader pair.
/// Different buckets and disks belonging to the same cycle share this entry,
/// so a TTL refresh cannot split one published cycle across generations.
pub(crate) async fn runtime_tier_registry_for_cycle(cycle: u64, leader_epoch: u64) -> TierRegistrySnapshot {
let key = (cycle, leader_epoch);
prune_inactive_tier_cycle_snapshots(cycle, leader_epoch);
{
let cached = TIER_CYCLE_SNAPSHOTS.read().unwrap_or_else(|err| err.into_inner());
if let Some(snapshot) = cached.get(&key) {
return snapshot.clone();
let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
if let Some((refreshed_at, names)) = cached
&& refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
{
return names;
}
}
let mut snapshot = runtime_tier_registry().await;
// The registry generation describes the configuration snapshot, not the
// scan that consumed it. Keep it stable across cycles so a healthy cache
// can be reused; cycle and leader fencing are carried separately by the
// cache metadata and scan plan.
snapshot.generation = tier_registry_generation(&snapshot.names);
let mut cached = TIER_CYCLE_SNAPSHOTS.write().unwrap_or_else(|err| err.into_inner());
if let Some(existing) = cached.get(&key) {
return existing.clone();
}
cached.insert(key, snapshot.clone());
snapshot
}
fn prune_inactive_tier_cycle_snapshots(cycle: u64, leader_epoch: u64) {
let active = TIER_ACTIVE_CYCLES.read().unwrap_or_else(|err| err.into_inner());
let mut snapshots = TIER_CYCLE_SNAPSHOTS.write().unwrap_or_else(|err| err.into_inner());
snapshots.retain(|(entry_cycle, entry_epoch), _| {
active.contains_key(&(*entry_cycle, *entry_epoch))
|| *entry_epoch > leader_epoch
|| (*entry_epoch == leader_epoch && *entry_cycle >= cycle)
});
}
pub(crate) struct TierRegistryCycleGuard {
key: (u64, u64),
}
impl Drop for TierRegistryCycleGuard {
fn drop(&mut self) {
let mut active = TIER_ACTIVE_CYCLES.write().unwrap_or_else(|err| err.into_inner());
if let Some(count) = active.get_mut(&self.key) {
*count = count.saturating_sub(1);
if *count == 0 {
active.remove(&self.key);
}
}
}
}
pub(crate) fn begin_tier_registry_cycle(cycle: u64, leader_epoch: u64) -> TierRegistryCycleGuard {
let mut active = TIER_ACTIVE_CYCLES.write().unwrap_or_else(|err| err.into_inner());
let count = active.entry((cycle, leader_epoch)).or_default();
*count = count.saturating_add(1);
TierRegistryCycleGuard {
key: (cycle, leader_epoch),
}
}
fn tier_registry_generation(names: &[String]) -> u64 {
let mut hasher = Sha256::new();
hasher.update(b"rustfs-tier-registry-v1");
for name in names {
hasher.update(u64::try_from(name.len()).unwrap_or(u64::MAX).to_le_bytes());
hasher.update(name.as_bytes());
}
let digest = hasher.finalize();
let mut prefix = [0_u8; 8];
prefix.copy_from_slice(&digest[..8]);
u64::from_le_bytes(prefix)
}
/// Drop cycle snapshots only after the scanner has finished publishing a
/// cycle. In-flight or retryable cycles must retain their original registry;
/// TTL/capacity eviction could make a later bucket in the same cycle refresh
/// to a different generation.
pub(crate) fn complete_tier_registry_cycle(cycle: u64, leader_epoch: u64) {
let active = TIER_ACTIVE_CYCLES.read().unwrap_or_else(|err| err.into_inner());
let mut cached = TIER_CYCLE_SNAPSHOTS.write().unwrap_or_else(|err| err.into_inner());
cached.retain(|(entry_cycle, entry_epoch), _| {
active.contains_key(&(*entry_cycle, *entry_epoch))
|| *entry_epoch > leader_epoch
|| (*entry_epoch == leader_epoch && *entry_cycle > cycle)
});
let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers();
let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::<Vec<_>>().into();
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names)));
names
}
/// Test-only cache reset; the production cache has no invalidation hook
@@ -619,9 +408,6 @@ pub(crate) fn complete_tier_registry_cycle(cycle: u64, leader_epoch: u64) {
#[cfg(test)]
fn reset_tier_name_cache_for_test() {
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = None;
TIER_ACTIVE_CYCLES.write().unwrap_or_else(|err| err.into_inner()).clear();
TIER_CYCLE_SNAPSHOTS.write().unwrap_or_else(|err| err.into_inner()).clear();
TIER_REGISTRY_GENERATION.store(0, Ordering::Relaxed);
}
pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
@@ -830,68 +616,6 @@ mod tests {
assert!(Arc::ptr_eq(&first, &second));
}
#[tokio::test]
async fn tier_registry_cycle_snapshot_stays_fixed_while_active() {
reset_tier_name_cache_for_test();
let cycle = 9_000_001;
let leader_epoch = 9_000_002;
let guard = begin_tier_registry_cycle(cycle, leader_epoch);
let first = runtime_tier_registry_for_cycle(cycle, leader_epoch).await;
// Simulate a TTL refresh observing a different configuration while the
// original cycle is still scanning. The active cycle entry must win.
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((
Instant::now() - TIER_NAME_CACHE_TTL - Duration::from_secs(1),
TierRegistrySnapshot {
generation: u64::MAX,
names: Arc::from(["COLD".to_string()]),
refresh_failed: false,
},
));
let second = runtime_tier_registry_for_cycle(cycle, leader_epoch).await;
assert_eq!(second.generation, first.generation);
assert_eq!(second.names, first.names);
drop(guard);
complete_tier_registry_cycle(cycle, leader_epoch);
reset_tier_name_cache_for_test();
}
#[tokio::test]
async fn tier_registry_generation_survives_new_cycle_with_same_names() {
reset_tier_name_cache_for_test();
let first_cycle = 9_000_011;
let second_cycle = first_cycle + 1;
let leader_epoch = 9_000_012;
let first_guard = begin_tier_registry_cycle(first_cycle, leader_epoch);
let first = runtime_tier_registry_for_cycle(first_cycle, leader_epoch).await;
drop(first_guard);
complete_tier_registry_cycle(first_cycle, leader_epoch);
let second_guard = begin_tier_registry_cycle(second_cycle, leader_epoch);
let second = runtime_tier_registry_for_cycle(second_cycle, leader_epoch).await;
assert_eq!(first.names, second.names);
assert_eq!(first.generation, second.generation);
drop(second_guard);
complete_tier_registry_cycle(second_cycle, leader_epoch);
reset_tier_name_cache_for_test();
}
#[test]
fn invalid_tier_registry_names_fail_closed_for_refresh() {
assert!(validate_tier_registry_names(vec!["COLD\n".to_string()]).is_err());
assert!(validate_tier_registry_names(vec![UNKNOWN_TIER.to_string()]).is_err());
assert!(validate_tier_registry_names(vec!["COLD".to_string(), "COLD".to_string()]).is_err());
assert_eq!(
validate_tier_registry_names(vec!["WARM".to_string(), "COLD".to_string()])
.expect("valid registry names")
.as_ref(),
["COLD".to_string(), "WARM".to_string()]
);
}
#[test]
fn foreground_read_guard_tracks_stream_lifetime() {
reset_foreground_read_activity_for_test();
+3 -36
View File
@@ -16,8 +16,8 @@
use crate::RUSTFS_META_BUCKET;
use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
use crate::scanner_io::{
DataUsageCacheReuseOptions, DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks,
cache_root_entry_info, current_cache_root_or_prepare_with_generation, scanner_set_disk_inventory,
DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
current_cache_root_or_prepare, scanner_set_disk_inventory,
};
use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
use crate::{
@@ -212,7 +212,6 @@ pub(crate) struct RemoteScannerScanSpec<'a> {
pub(crate) session_id: Uuid,
pub(crate) session_sequence: u64,
pub(crate) scan_plan_digest: DataUsageScanPlanDigest,
pub(crate) tier_registry_generation: u64,
pub(crate) skip_healing: bool,
pub(crate) scan_mode: HealScanMode,
}
@@ -223,7 +222,6 @@ struct RemoteScannerResponseExpectation<'a> {
source: DataUsageCacheSource,
next_cycle: u64,
scan_plan_digest: DataUsageScanPlanDigest,
tier_registry_generation: u64,
}
#[derive(Debug)]
@@ -670,11 +668,6 @@ async fn scan_and_persist_local_bucket(
scan_mode,
..
} = request;
// Keep the worker's cycle snapshot alive through cache reuse, scanning,
// and persistence. Without the guard, a later cycle can prune this key
// while this request is still running and allow a second registry to be
// selected for the same cycle.
let _tier_cycle_guard = crate::begin_tier_registry_cycle(next_cycle, leader_epoch);
let store = resolve_scanner_object_store_handle()
.ok_or_else(|| RemoteScannerServerError::worker("remote namespace scanner object layer is unavailable"))?;
validate_remote_scanner_request_fence_with_store(next_cycle, leader_epoch, store.clone())
@@ -707,25 +700,7 @@ async fn scan_and_persist_local_bucket(
let revisions = cache.load_with_revisions(set.clone(), &cache_name).await.map_err(|err| {
RemoteScannerServerError::worker(format!("remote namespace scanner cache load or revision lookup failed: {err}"))
})?;
// Remote workers use the same cycle-frozen registry as `scan_data_folder`.
// Requiring its generation here prevents a cache snapshot classified by an
// older registry from being reused before the folder scan gets a chance to
// refresh it.
let tier_registry_generation = crate::runtime_tier_registry_for_cycle(next_cycle, leader_epoch)
.await
.generation;
let scan_state = current_cache_root_or_prepare_with_generation(
&mut cache,
&bucket,
source,
next_cycle,
leader_epoch,
scan_plan_digest,
DataUsageCacheReuseOptions {
require_source: true,
tier_registry_generation: Some(tier_registry_generation),
},
);
let scan_state = current_cache_root_or_prepare(&mut cache, &bucket, source, next_cycle, leader_epoch, scan_plan_digest, true);
match scan_state {
DataUsageCacheScanState::Current(usage) => {
if guard.is_lock_lost() {
@@ -869,7 +844,6 @@ pub(crate) async fn scan_remote_bucket(
session_id,
session_sequence,
scan_plan_digest,
tier_registry_generation,
skip_healing,
scan_mode,
} = spec;
@@ -958,7 +932,6 @@ pub(crate) async fn scan_remote_bucket(
source: expected_source,
next_cycle,
scan_plan_digest,
tier_registry_generation,
},
authenticator,
rpc_deadline,
@@ -1014,7 +987,6 @@ where
source: expected_source,
next_cycle: TEST_NEXT_CYCLE,
scan_plan_digest: expected_scan_plan_digest,
tier_registry_generation: 0,
},
authenticator,
Instant::now() + NS_SCANNER_MAX_RPC_LIFETIME,
@@ -1114,11 +1086,6 @@ where
"remote namespace scanner returned usage for a different bucket plan",
)));
}
if complete.usage.tier_registry_generation != Some(expected.tier_registry_generation) {
return Err(RemoteScannerStreamError::reconciled(StorageError::other(
"remote namespace scanner returned usage for a different tier registry generation",
)));
}
if !complete.usage.entry.children.is_empty() {
return Err(RemoteScannerStreamError::reconciled(StorageError::other(
"remote namespace scanner returned non-flattened bucket usage",
@@ -195,7 +195,6 @@ fn test_usage(bucket: &str, objects: usize) -> DataUsageEntryInfo {
name: bucket.to_string(),
parent: crate::DATA_USAGE_ROOT.to_string(),
entry,
tier_registry_generation: Some(0),
}
}
@@ -752,43 +751,6 @@ async fn complete_terminal_frame_reconciles_progress_and_usage() {
assert_eq!(budget.progress(), (3, 2));
}
#[tokio::test]
async fn terminal_usage_from_a_different_tier_generation_is_rejected() {
let request_id = Uuid::new_v4();
let writer_auth = FrameAuthenticator::for_test(request_id);
let reader_auth = FrameAuthenticator::for_test(request_id);
let (mut writer, reader) = tokio::io::duplex(4096);
tokio::spawn(async move {
let mut usage = test_usage("bucket", 1);
usage.tier_registry_generation = Some(1);
let mut sequence = 0;
write_frame(
&mut writer,
&writer_auth,
&mut sequence,
&RemoteScannerFrame::terminal(
RemoteScannerProgress::default(),
RemoteScannerFrameResult::Complete(Box::new(RemoteScannerComplete {
source: TEST_SOURCE,
scan_plan_digest: TEST_PLAN_DIGEST,
usage,
pending_maintenance_work: false,
})),
),
)
.await
.expect("terminal frame should write");
});
let parent = CancellationToken::new();
let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
let error = consume_remote_scanner_stream(reader, parent, budget, "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth)
.await
.expect_err("generation mismatch must fail closed");
assert!(error.to_string().contains("tier registry generation"));
}
#[tokio::test]
async fn complete_terminal_frame_after_budget_expiry_is_partial() {
let request_id = Uuid::new_v4();
+4 -32
View File
@@ -55,9 +55,9 @@ use tracing::{debug, error, warn};
use crate::{
Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig,
ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _,
ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, TierRegistrySnapshot, apply_expiry_rule,
apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal, runtime_tier_registry_for_cycle, scanner_is_erasure,
ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure,
scanner_replication_config_for_lifecycle_eval,
};
use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
@@ -626,20 +626,6 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
}
into.add_tier_sizes(&summary.tier_stats);
into.add_unknown_tier_stats(&summary.unknown_tier_stats);
into.tier_accounting_proof = match (into.tier_accounting_proof, Some(summary.tier_accounting_proof)) {
(Some(mut current), Some(next)) => {
current.saturating_add(next);
Some(current)
}
(Some(_), None) => None,
(None, next) => next,
};
if into.unknown_tier_stats.as_ref().is_some_and(|stats| stats.counter_overflowed)
&& let Some(proof) = into.tier_accounting_proof.as_mut()
{
proof.overflowed = true;
}
}
fn data_usage_root_has_progress(root: &DataUsageEntry) -> bool {
@@ -650,8 +636,6 @@ fn data_usage_root_has_progress(root: &DataUsageEntry) -> bool {
|| root.delete_markers > 0
|| root.failed_objects > 0
|| root.replication_stats.is_some()
|| root.all_tier_stats.as_ref().is_some_and(|stats| !stats.is_empty())
|| root.unknown_tier_stats.as_ref().is_some_and(|stats| !stats.is_empty())
}
fn partial_cache_is_useful(root: &DataUsageEntry, pending_heals_changed: bool) -> bool {
@@ -686,9 +670,6 @@ pub struct FolderScanner {
budget: Arc<ScannerCycleBudget>,
skip_heal: Arc<std::sync::atomic::AtomicBool>,
local_disk: Arc<Disk>,
/// Tier registry frozen for this folder scan. A refresh applies to the
/// next scan and cannot mix generations in one aggregate.
tier_registry: TierRegistrySnapshot,
pending_heals_changed: bool,
#[cfg(test)]
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
@@ -1348,11 +1329,7 @@ impl FolderScanner {
continue;
}
let sz = match self
.local_disk
.get_size_with_tier_names(item.clone(), &self.tier_registry.names)
.await
{
let sz = match self.local_disk.get_size(item.clone()).await {
Ok(sz) => sz,
Err(e) => {
let failure_action = classify_get_size_failure(&item, &e);
@@ -2184,10 +2161,6 @@ pub async fn scan_data_folder(
let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64;
let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize;
let tier_registry = runtime_tier_registry_for_cycle(cache.info.next_cycle, cache.info.leader_epoch).await;
let mut cache = cache;
cache.fold_retired_tiers(&tier_registry.names);
cache.info.tier_registry_generation = Some(tier_registry.generation);
// Create folder scanner
let mut scanner = FolderScanner {
@@ -2216,7 +2189,6 @@ pub async fn scan_data_folder(
budget: budget.clone(),
skip_heal,
local_disk,
tier_registry,
pending_heals_changed: false,
#[cfg(test)]
list_path_raw_options_observer: None,
@@ -289,62 +289,11 @@ impl ScannerItem {
item.object_path()
}
fn effective_tier(oi: &ObjectInfo) -> &str {
if oi.transitioned_object.status == crate::TRANSITION_COMPLETE {
oi.transitioned_object.tier.as_str()
} else {
oi.storage_class.as_deref().unwrap_or(crate::storageclass::STANDARD)
}
}
fn tier_name_is_known(tier: &str, tier_names: &[String]) -> bool {
!tier.is_empty()
&& tier != crate::data_usage_define::UNKNOWN_TIER
&& (tier == crate::storageclass::STANDARD
|| tier == crate::storageclass::RRS
|| tier_names.iter().any(|name| name == tier))
}
pub(crate) fn tier_is_known(oi: &ObjectInfo, tier_names: &[String]) -> bool {
Self::tier_name_is_known(Self::effective_tier(oi), tier_names)
}
fn action_requires_known_tier(action: IlmAction) -> bool {
matches!(
action,
IlmAction::TransitionAction
| IlmAction::TransitionVersionAction
| IlmAction::DeleteAction
| IlmAction::DeleteVersionAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction
)
}
fn action_blocked_by_unknown_tier(
action: IlmAction,
oi: &ObjectInfo,
all_versions_known: bool,
tier_names: &[String],
target: &str,
) -> bool {
if !Self::action_requires_known_tier(action) {
return false;
}
!Self::tier_is_known(oi, tier_names)
|| (action.delete_all() && !all_versions_known)
|| (matches!(action, IlmAction::TransitionAction | IlmAction::TransitionVersionAction)
&& !Self::tier_name_is_known(target, tier_names))
}
pub async fn apply_actions(
&mut self,
object_infos: Vec<ObjectInfo>,
lock_retention: Option<Arc<ObjectLockConfiguration>>,
versioning_config: VersioningConfiguration,
tier_names: &[String],
size_summary: &mut SizeSummary,
) {
let object_path = self.object_path();
@@ -453,9 +402,6 @@ impl ScannerItem {
let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new();
let mut cumulative_size = 0;
let mut remaining_versions = object_infos.len();
let all_versions_known = object_infos
.iter()
.all(|candidate| Self::tier_is_known(candidate, tier_names));
'eventLoop: {
for (i, event) in events.iter().enumerate() {
let oi = &object_infos[i];
@@ -479,18 +425,6 @@ impl ScannerItem {
let mut size = actual_size;
let mut account_now = true;
// A retired/unknown source tier may point at a remote object
// that cannot be safely deleted or transitioned. Lifecycle
// evaluation is still useful for accounting, but all
// side-effecting tier actions fail closed until the registry
// recognizes the source again.
if Self::action_blocked_by_unknown_tier(event.action, oi, all_versions_known, tier_names, &event.storage_class) {
size = self.heal_actions(oi, actual_size, size_summary).await;
size_summary.actions_accounting(oi, size, actual_size);
cumulative_size += size;
continue;
}
match event.action {
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
debug!(
@@ -995,49 +929,4 @@ mod tests {
assert_eq!(item.object_name, "object");
assert_eq!(item.object_path(), "object");
}
#[test]
fn unknown_tier_never_triggers_transition() {
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
..Default::default()
};
let tier_names = ["WARM".to_string()];
assert!(!ScannerItem::tier_is_known(&object, &tier_names));
assert!(ScannerItem::action_requires_known_tier(IlmAction::TransitionAction));
assert!(ScannerItem::action_requires_known_tier(IlmAction::DeleteVersionAction));
assert!(ScannerItem::action_blocked_by_unknown_tier(
IlmAction::TransitionAction,
&object,
false,
&tier_names,
"WARM"
));
assert!(!ScannerItem::action_blocked_by_unknown_tier(
IlmAction::NoneAction,
&object,
false,
&tier_names,
"WARM"
));
let known = ObjectInfo {
storage_class: Some(crate::storageclass::STANDARD.to_string()),
..Default::default()
};
assert!(ScannerItem::action_blocked_by_unknown_tier(
IlmAction::DeleteAllVersionsAction,
&known,
false,
&tier_names,
"WARM"
));
assert!(!ScannerItem::action_blocked_by_unknown_tier(
IlmAction::TransitionAction,
&known,
true,
&tier_names,
crate::storageclass::STANDARD
));
}
}
@@ -325,11 +325,6 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
budget: ScannerCycleBudget::new(&CancellationToken::new(), Default::default()),
skip_heal: Arc::new(AtomicBool::new(false)),
local_disk: disk,
tier_registry: crate::TierRegistrySnapshot {
generation: 0,
names: Arc::new([]),
refresh_failed: false,
},
pending_heals_changed: false,
list_path_raw_options_observer: None,
};
+4 -20
View File
@@ -57,9 +57,8 @@ use crate::storage_api::scanner_io::{BucketInfo, BucketOptions};
use crate::{
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, begin_tier_registry_cycle,
complete_tier_registry_cycle, enqueue_runtime_free_version, get_lifecycle_config, get_object_lock_config,
get_replication_config, runtime_tier_names, runtime_tier_registry_for_cycle, storageclass,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, runtime_tier_names, storageclass,
};
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
@@ -144,7 +143,6 @@ pub struct ScannerBucketScanPlan {
all_buckets: Arc<Vec<BucketInfo>>,
digest: DataUsageScanPlanDigest,
leader_epoch: u64,
tier_registry_generation: u64,
dirty_usage_buckets: Arc<DirtyUsageBuckets>,
bucket_failures: ScannerBucketFailureState,
pending_maintenance_work: Arc<AtomicBool>,
@@ -343,20 +341,12 @@ pub(crate) fn cache_root_entry_info(cache: &DataUsageCache) -> std::result::Resu
name: cache.info.name.clone(),
parent: DATA_USAGE_ROOT.to_string(),
entry,
tier_registry_generation: cache.info.tier_registry_generation,
})
}
fn apply_bucket_result_to_cache(cache: &mut DataUsageCache, result: DataUsageEntryInfo, update_time: SystemTime) -> bool {
if cache.info.tier_registry_generation != result.tier_registry_generation {
// A result from another registry generation must never be folded into
// this cycle. Leaving it unapplied makes the cycle incomplete and
// forces the caller to re-account it under one frozen registry.
return false;
}
fn apply_bucket_result_to_cache(cache: &mut DataUsageCache, result: DataUsageEntryInfo, update_time: SystemTime) {
cache.replace(&result.name, &result.parent, result.entry);
cache.info.last_update = Some(update_time);
true
}
fn should_publish_completed_snapshot(completed_count: usize, total_count: usize, budget_elapsed: bool, cancelled: bool) -> bool {
@@ -508,9 +498,6 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static {
) -> Result<ScannerDiskScanOutcome>;
async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary>;
/// Read one object using a registry snapshot captured at scan start.
async fn get_size_with_tier_names(&self, item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary>;
}
#[derive(Debug)]
@@ -662,10 +649,7 @@ use cache::*;
use dirty_usage::*;
use guards::*;
pub(crate) use cache::{
DataUsageCacheReuseOptions, DataUsageCacheScanState, acquire_scanner_cache_locks,
current_cache_root_or_prepare_with_generation,
};
pub(crate) use cache::{DataUsageCacheScanState, acquire_scanner_cache_locks, current_cache_root_or_prepare};
pub use dirty_usage::{
ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,
+6 -181
View File
@@ -100,14 +100,13 @@ where
let _ = tokio::time::timeout(SCANNER_CACHE_LOCK_LOSS_SHUTDOWN_TIMEOUT, scan).await;
}
pub(crate) fn current_cache_root_entry_with_generation(
pub(crate) fn current_cache_root_entry(
cache: &DataUsageCache,
name: &str,
source: DataUsageCacheSource,
next_cycle: u64,
leader_epoch: u64,
scan_plan_digest: DataUsageScanPlanDigest,
tier_registry_generation: Option<u64>,
) -> std::result::Result<Option<DataUsageEntryInfo>, ScannerError> {
let metadata_is_current = cache.info.name == name
&& cache.info.source == Some(source)
@@ -116,8 +115,7 @@ pub(crate) fn current_cache_root_entry_with_generation(
&& cache.info.last_update.is_some()
&& cache.info.next_cycle == next_cycle
&& cache.info.leader_epoch == leader_epoch
&& cache.info.cache_key_format == DATA_USAGE_CACHE_KEY_FORMAT
&& tier_registry_generation.is_none_or(|generation| cache.info.tier_registry_generation == Some(generation));
&& cache.info.cache_key_format == DATA_USAGE_CACHE_KEY_FORMAT;
if !metadata_is_current {
return Ok(None);
}
@@ -133,13 +131,6 @@ pub(crate) enum DataUsageCacheScanState {
},
}
#[derive(Clone, Copy, Debug, Default)]
pub(crate) struct DataUsageCacheReuseOptions {
pub(crate) require_source: bool,
pub(crate) tier_registry_generation: Option<u64>,
}
#[cfg(test)]
pub(crate) fn current_cache_root_or_prepare(
cache: &mut DataUsageCache,
name: &str,
@@ -149,51 +140,11 @@ pub(crate) fn current_cache_root_or_prepare(
scan_plan_digest: DataUsageScanPlanDigest,
require_source: bool,
) -> DataUsageCacheScanState {
current_cache_root_or_prepare_with_generation(
cache,
name,
source,
next_cycle,
leader_epoch,
scan_plan_digest,
DataUsageCacheReuseOptions {
require_source,
tier_registry_generation: None,
},
)
}
pub(crate) fn current_cache_root_or_prepare_with_generation(
cache: &mut DataUsageCache,
name: &str,
source: DataUsageCacheSource,
next_cycle: u64,
leader_epoch: u64,
scan_plan_digest: DataUsageScanPlanDigest,
options: DataUsageCacheReuseOptions,
) -> DataUsageCacheScanState {
if options.tier_registry_generation.is_some_and(|generation| {
cache.info.next_cycle <= next_cycle
&& cache.info.leader_epoch <= leader_epoch
&& cache.info.tier_registry_generation != Some(generation)
}) {
// Make prepare_for_scan take its reset path so an entry classified by
// an older registry cannot be reused under the new cycle generation.
cache.info.scan_plan_digest = None;
}
match current_cache_root_entry_with_generation(
cache,
name,
source,
next_cycle,
leader_epoch,
scan_plan_digest,
options.tier_registry_generation,
) {
match current_cache_root_entry(cache, name, source, next_cycle, leader_epoch, scan_plan_digest) {
Ok(Some(root)) => DataUsageCacheScanState::Current(Box::new(root)),
current => DataUsageCacheScanState::Prepared {
invalid_current: current.err(),
outcome: cache.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, options.require_source),
outcome: cache.prepare_for_scan(name, next_cycle, leader_epoch, source, scan_plan_digest, require_source),
},
}
}
@@ -208,7 +159,7 @@ pub(super) fn cache_snapshot_is_current(
scan_plan_digest: DataUsageScanPlanDigest,
) -> bool {
matches!(
current_cache_root_entry_with_generation(cache, name, source, next_cycle, leader_epoch, scan_plan_digest, None),
current_cache_root_entry(cache, name, source, next_cycle, leader_epoch, scan_plan_digest),
Ok(Some(_))
)
}
@@ -217,7 +168,6 @@ pub(super) fn completed_data_usage_info(
results: &[DataUsageCache],
expected_sources: &HashSet<DataUsageCacheSource>,
all_buckets: &[String],
tier_registry_names: &[String],
bucket_plan_complete: bool,
budget_elapsed: bool,
cancelled: bool,
@@ -233,19 +183,6 @@ pub(super) fn completed_data_usage_info(
return None;
}
// A generation is comparable across nodes because it is derived from the
// frozen registry names. Cycle and leader fencing remain separate cache
// metadata. Legacy peers omit the generation; an all-legacy result remains
// readable, but mixing legacy and new (or two new generations) would make
// the per-tier accounting ambiguous.
let registry_generation = results.first()?.info.tier_registry_generation;
if results.iter().any(|result| match registry_generation {
Some(generation) => result.info.tier_registry_generation != Some(generation),
None => result.info.tier_registry_generation.is_some(),
}) {
return None;
}
if results.iter().any(|result| result.root().is_none()) {
return None;
}
@@ -263,16 +200,9 @@ pub(super) fn completed_data_usage_info(
if !total.checked_merge(&merged) {
return None;
}
if !tier_accounting_proof_is_publishable(&merged, registry_generation, tier_registry_names) {
return None;
}
buckets_usage.insert(bucket.clone(), checked_bucket_usage_info(&merged)?);
}
if !tier_accounting_proof_is_publishable(&total, registry_generation, tier_registry_names) {
return None;
}
let merged_last_update = results.iter().filter_map(|result| result.info.last_update).max()?;
let bucket_sizes = buckets_usage
.iter()
@@ -286,7 +216,6 @@ pub(super) fn completed_data_usage_info(
delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
objects_total_size: u64::try_from(total.size).ok()?,
tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: total.unknown_tier_stats.filter(|stats| !stats.is_empty()),
buckets_count: u64::try_from(all_buckets.len()).ok()?,
bucket_sizes,
buckets_usage,
@@ -296,109 +225,6 @@ pub(super) fn completed_data_usage_info(
Some((data_usage_info, merged_last_update))
}
fn tier_accounting_proof_is_publishable(
entry: &DataUsageEntry,
registry_generation: Option<u64>,
tier_registry_names: &[String],
) -> bool {
let has_scalar_usage = entry.size > 0
|| entry.objects > 0
|| entry.versions > 0
|| entry.delete_markers > 0
|| entry.failed_objects > 0
|| !entry.obj_sizes.is_empty()
|| !entry.obj_versions.is_empty()
|| entry.replication_stats.as_ref().is_some_and(|stats| !stats.is_empty());
let has_tier_accounted_data = entry
.all_tier_stats
.as_ref()
.is_some_and(|stats| stats.tiers.values().any(|tier| !tier.is_empty()))
|| entry.unknown_tier_stats.as_ref().is_some_and(|stats| {
stats.counter_overflowed
|| stats.unknown_bytes > 0
|| stats.unknown_physical_bytes > 0
|| stats.unknown_objects > 0
|| stats.unknown_versions > 0
});
let Some(proof) = entry.tier_accounting_proof else {
return !has_scalar_usage && !has_tier_accounted_data;
};
if proof.overflowed
|| entry
.unknown_tier_stats
.as_ref()
.is_some_and(|stats| stats.counter_overflowed)
|| u64::try_from(entry.size).ok() != Some(proof.logical_total)
{
return false;
}
let unknown_logical = entry.unknown_tier_stats.as_ref().map_or(0, |stats| stats.unknown_bytes);
let unknown_physical = entry
.unknown_tier_stats
.as_ref()
.map_or(0, |stats| stats.unknown_physical_bytes);
if proof
.logical_known
.checked_add(unknown_logical)
.is_none_or(|total| total != proof.logical_total)
|| proof
.physical_known
.checked_add(unknown_physical)
.is_none_or(|total| total != proof.physical_total)
{
return false;
}
if registry_generation.is_some()
&& entry.all_tier_stats.as_ref().is_some_and(|stats| {
stats.tiers.keys().any(|tier| {
tier != crate::UNKNOWN_TIER
&& tier != crate::storageclass::STANDARD
&& tier != crate::storageclass::RRS
&& !tier_registry_names.iter().any(|allowed| allowed == tier)
})
})
{
return false;
}
if !has_tier_accounted_data {
return true;
}
let Some(tiers) = entry.all_tier_stats.as_ref() else {
return false;
};
let map_unknown_physical = tiers.tiers.get(crate::UNKNOWN_TIER).map_or(0, |stats| stats.total_size);
let companion_unknown_physical = entry
.unknown_tier_stats
.as_ref()
.map_or(0, |stats| stats.unknown_physical_bytes);
if map_unknown_physical != companion_unknown_physical {
return false;
}
// A no-configuration scan intentionally stores only UNKNOWN_TIER after
// the first unknown object; STANDARD/RRS remain absent to preserve the
// historical empty-map shape. In that shape the scalar proof is the sole
// source of known physical bytes. Configured registries seed at least one
// non-UNKNOWN key, whose map total must match the proof.
let has_known_tier_map = tiers.tiers.keys().any(|tier| tier.as_str() != crate::UNKNOWN_TIER);
if !has_known_tier_map {
return true;
}
let Some(known_tier_physical_total) = tiers
.tiers
.iter()
.filter(|(tier, _)| tier.as_str() != crate::UNKNOWN_TIER)
.map(|(_, stats)| stats)
.try_fold(0_u64, |total, stats| total.checked_add(stats.total_size))
else {
return false;
};
proof.physical_known == known_tier_physical_total
}
pub(super) async fn send_cache_root_entry_info(
bucket_result_tx: &mpsc::Sender<DataUsageEntryInfo>,
cache: &DataUsageCache,
@@ -493,14 +319,13 @@ pub(super) async fn persist_and_publish_cache_snapshot(
return None;
}
if matches!(
current_cache_root_entry_with_generation(
current_cache_root_entry(
&persisted,
DATA_USAGE_ROOT,
source,
cache_snapshot.info.next_cycle,
cache_snapshot.info.leader_epoch,
scan_plan_digest,
cache_snapshot.info.tier_registry_generation,
),
Ok(Some(_))
) {
+4 -20
View File
@@ -31,7 +31,6 @@ impl ScannerIOCache for SetDisks {
all_buckets,
digest: scan_plan_digest,
leader_epoch,
tier_registry_generation,
dirty_usage_buckets,
bucket_failures,
pending_maintenance_work,
@@ -49,7 +48,6 @@ impl ScannerIOCache for SetDisks {
next_cycle: want_cycle,
last_update: Some(now),
leader_epoch,
tier_registry_generation: Some(tier_registry_generation),
source: Some(source),
snapshot_complete: true,
scan_plan_digest: Some(scan_plan_digest),
@@ -220,13 +218,6 @@ impl ScannerIOCache for SetDisks {
"Scanner old data usage cache load failed; rebuilding from bucket caches"
);
}
// Fence a stale set aggregate before copying entries into per-bucket work caches.
if old_cache.info.next_cycle <= want_cycle
&& old_cache.info.leader_epoch <= leader_epoch
&& old_cache.info.tier_registry_generation != Some(tier_registry_generation)
{
old_cache.info.scan_plan_digest = None;
}
match old_cache.prepare_for_scan(
DATA_USAGE_ROOT,
want_cycle,
@@ -276,7 +267,6 @@ impl ScannerIOCache for SetDisks {
name: DATA_USAGE_ROOT.to_string(),
next_cycle: want_cycle,
leader_epoch,
tier_registry_generation: Some(tier_registry_generation),
source: Some(source),
snapshot_complete: false,
scan_plan_digest: Some(scan_plan_digest),
@@ -338,9 +328,8 @@ impl ScannerIOCache for SetDisks {
};
let mut cache = cache_mutex_clone.lock().await;
if apply_bucket_result_to_cache(&mut cache, result, SystemTime::now()) {
completed_bucket_count_clone.fetch_add(1, Ordering::Relaxed);
}
apply_bucket_result_to_cache(&mut cache, result, SystemTime::now());
completed_bucket_count_clone.fetch_add(1, Ordering::Relaxed);
}
}
}
@@ -471,7 +460,6 @@ impl ScannerIOCache for SetDisks {
session_id: remote_session_id,
session_sequence: request_sequence,
scan_plan_digest: bucket_scan_plan_digest,
tier_registry_generation,
skip_healing: healing,
scan_mode,
},
@@ -687,17 +675,14 @@ impl ScannerIOCache for SetDisks {
continue;
}
};
let scan_state = current_cache_root_or_prepare_with_generation(
let scan_state = current_cache_root_or_prepare(
&mut cache,
&bucket.name,
source,
want_cycle,
leader_epoch,
bucket_scan_plan_digest,
DataUsageCacheReuseOptions {
require_source: require_cache_source,
tier_registry_generation: Some(tier_registry_generation),
},
require_cache_source,
);
let outcome = match scan_state {
DataUsageCacheScanState::Current(root) => {
@@ -1123,7 +1108,6 @@ impl ScannerIOCache for SetDisks {
name: DATA_USAGE_ROOT.to_string(),
next_cycle: want_cycle,
leader_epoch,
tier_registry_generation: Some(tier_registry_generation),
source: Some(source),
snapshot_complete: false,
scan_plan_digest: Some(scan_plan_digest),
-11
View File
@@ -48,7 +48,6 @@ impl ScannerIOCycle for ECStore {
scan_mode: HealScanMode,
) -> Result<ScannerCycleResult> {
let child_token = ctx.child_token();
let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
// Check the local pool metadata before listing buckets. A failed or
// canceled decommission remains suspended after its worker exits, so
@@ -128,8 +127,6 @@ impl ScannerIOCycle for ECStore {
scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before));
let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
let tier_registry_generation = tier_registry.generation;
if all_buckets.is_empty() {
reset_set_scan_gauges();
@@ -160,9 +157,6 @@ impl ScannerIOCycle for ECStore {
{
return Ok(ScannerCycleResult::new(status, None));
}
if status == ScannerCycleStatus::Complete {
complete_tier_registry_cycle(want_cycle, leader_epoch);
}
let dirty_usage_clear =
(status == ScannerCycleStatus::Complete).then(|| dirty_usage_snapshot.buckets.as_ref().clone());
let remote_dirty_usage_acknowledgements = if status == ScannerCycleStatus::Complete {
@@ -255,7 +249,6 @@ impl ScannerIOCycle for ECStore {
all_buckets: Arc::clone(&all_buckets),
digest: scan_plan_digest,
leader_epoch,
tier_registry_generation,
dirty_usage_buckets: dirty_usage_snapshot.buckets.clone(),
bucket_failures: bucket_failures.clone(),
pending_maintenance_work: pending_maintenance_work.clone(),
@@ -373,7 +366,6 @@ impl ScannerIOCycle for ECStore {
&results,
&expected_sources,
&all_bucket_names,
&tier_registry.names,
bucket_plan_complete,
budget_elapsed,
ctx.is_cancelled(),
@@ -399,9 +391,6 @@ impl ScannerIOCycle for ECStore {
&failed_buckets,
);
result?;
if cycle_status == ScannerCycleStatus::Complete {
complete_tier_registry_cycle(want_cycle, leader_epoch);
}
let remote_dirty_usage_acknowledgements = if cycle_status == ScannerCycleStatus::Complete {
crate::scanner::scanner_dirty_usage_acknowledgements(&activity_before)
} else {
+15 -27
View File
@@ -13,38 +13,29 @@
// limitations under the License.
/// ScannerIODisk implementation for Disk: get_size and the per-disk bucket scan.
use super::*;
use crate::UNKNOWN_TIER;
///
/// Seed [`SizeSummary::tier_stats`] from the cached tier-name list.
///
/// Preserves the original no-tier shape: with no tiers configured the map
/// stays completely empty (STANDARD/RRS/UNKNOWN are not seeded either).
/// Otherwise the standard storage classes and one fixed unknown bucket are
/// seeded alongside every configured tier so per-object accounting never
/// inserts an untrusted metadata key.
/// Preserves the original seeding semantics: with no tiers configured the map
/// stays completely empty (STANDARD/RRS are not seeded either); otherwise the
/// standard storage classes are seeded alongside every configured tier so
/// per-object accounting always finds its tier key.
pub(super) fn tier_stats_template(tier_names: &[String]) -> HashMap<String, TierStats> {
let mut tier_stats = HashMap::with_capacity(tier_names.len() + 3);
let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2);
for tier_name in tier_names {
if tier_name != UNKNOWN_TIER {
tier_stats.insert(tier_name.clone(), TierStats::default());
}
tier_stats.insert(tier_name.clone(), TierStats::default());
}
if !tier_stats.is_empty() {
tier_stats.insert(storageclass::STANDARD.to_string(), TierStats::default());
tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
tier_stats.insert(UNKNOWN_TIER.to_string(), TierStats::default());
}
tier_stats
}
#[async_trait::async_trait]
impl ScannerIODisk for Disk {
async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary> {
self.get_size_with_tier_names(item, &runtime_tier_names().await).await
}
async fn get_size_with_tier_names(&self, mut item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary> {
async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> {
let done_object = Metrics::time(Metric::ScanObject);
if !is_xl_meta_path(&item.path) {
@@ -114,13 +105,12 @@ impl ScannerIODisk for Disk {
.map(|v| ObjectInfo::from_file_info(v, item.bucket.as_str(), object_path.as_str(), versioned))
.collect::<Vec<ObjectInfo>>();
// The caller supplies one registry snapshot for the whole folder scan;
// seeding from it prevents a TTL refresh from mixing generations in a
// single result.
let mut size_summary = SizeSummary {
tier_stats: tier_stats_template(tier_names),
..Default::default()
};
let mut size_summary = SizeSummary::default();
// Tier names come from the process-wide TTL cache; seeding from them
// replaces the per-object clone of every full TierConfig.
let tier_names = runtime_tier_names().await;
size_summary.tier_stats = tier_stats_template(&tier_names);
let lock_config = object_lock_config_for_scanner_item(&item).await;
@@ -130,14 +120,12 @@ impl ScannerIODisk for Disk {
// `object_infos`.
global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
item.apply_actions(object_infos, lock_config, versioning_config, tier_names, &mut size_summary)
item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary)
.await;
if !free_version_infos.is_empty() {
for oi in free_version_infos {
if ScannerItem::tier_is_known(&oi, tier_names) {
enqueue_runtime_free_version(oi).await;
}
enqueue_runtime_free_version(oi).await;
}
}
@@ -13,8 +13,7 @@
// limitations under the License.
use super::*;
use crate::data_usage_define::{UNKNOWN_TIER, UnknownTierStats, hash_path};
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage, TierAccountingProof};
use rustfs_data_usage::{ReplicationAllStats, ReplicationTargetUsage};
const TEST_PLAN_DIGEST: DataUsageScanPlanDigest = DataUsageScanPlanDigest([7; 32]);
@@ -90,11 +89,6 @@ fn completed_root_cache(bucket: &str, objects: usize, update_secs: u64, source:
DataUsageEntry {
objects,
size: objects.saturating_mul(10),
tier_accounting_proof: Some(TierAccountingProof {
logical_total: u64::try_from(objects.saturating_mul(10)).unwrap_or(u64::MAX),
logical_known: u64::try_from(objects.saturating_mul(10)).unwrap_or(u64::MAX),
..Default::default()
}),
..Default::default()
},
);
@@ -108,7 +102,7 @@ fn completed_data_usage_info_for_test(
cancelled: bool,
) -> Option<(DataUsageInfo, SystemTime)> {
let expected_sources = results.iter().filter_map(|result| result.info.source).collect::<HashSet<_>>();
completed_data_usage_info(results, &expected_sources, all_buckets, &[], true, budget_elapsed, cancelled)
completed_data_usage_info(results, &expected_sources, all_buckets, true, budget_elapsed, cancelled)
}
#[test]
@@ -128,21 +122,11 @@ fn completed_data_usage_info_publishes_tier_stats_across_sets() {
let mut first_set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let mut tiered = DataUsageEntry::default();
tiered.add_tier_sizes(&warm(100, 2, 1));
tiered.tier_accounting_proof = Some(TierAccountingProof {
physical_total: 100,
physical_known: 100,
..Default::default()
});
first_set.replace("bucket-b", DATA_USAGE_ROOT, tiered);
let mut second_set = completed_root_cache("bucket-b", 2, 20, DataUsageCacheSource::new(1, 0));
let mut tiered = DataUsageEntry::default();
tiered.add_tier_sizes(&warm(50, 1, 1));
tiered.tier_accounting_proof = Some(TierAccountingProof {
physical_total: 50,
physical_known: 50,
..Default::default()
});
second_set.replace("bucket-a", DATA_USAGE_ROOT, tiered);
let (data_usage_info, _) = completed_data_usage_info_for_test(&[first_set, second_set], &all_buckets, false, false)
@@ -161,269 +145,6 @@ fn completed_data_usage_info_publishes_tier_stats_across_sets() {
);
}
#[test]
fn completed_data_usage_info_rejects_logical_proof_mismatch() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_tier_sizes(&HashMap::from([(
"WARM".to_string(),
TierStats {
total_size: 10,
num_versions: 1,
num_objects: 1,
},
)]));
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 10,
logical_known: 9,
physical_total: 10,
physical_known: 10,
..Default::default()
});
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_rejects_logical_total_size_mismatch() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.size = 11;
entry.add_tier_sizes(&HashMap::from([(
"WARM".to_string(),
TierStats {
total_size: 10,
num_versions: 1,
num_objects: 1,
},
)]));
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 10,
logical_known: 10,
physical_total: 10,
physical_known: 10,
..Default::default()
});
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_rejects_physical_proof_mismatch() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_tier_sizes(&HashMap::from([(
"WARM".to_string(),
TierStats {
total_size: 10,
num_versions: 1,
num_objects: 1,
},
)]));
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 10,
logical_known: 10,
physical_total: 9,
physical_known: 9,
..Default::default()
});
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_rejects_unknown_physical_double_accounting() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_tier_sizes(&HashMap::from([(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: 10,
num_versions: 1,
num_objects: 1,
},
)]));
entry.add_unknown_tier_stats(&UnknownTierStats {
unknown_physical_bytes: 9,
..Default::default()
});
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 10,
logical_known: 10,
physical_total: 10,
physical_known: 10,
..Default::default()
});
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_rejects_unknown_counter_overflow() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_unknown_tier_stats(&UnknownTierStats {
counter_overflowed: true,
..Default::default()
});
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_accepts_no_tier_standard_empty_map_with_proof() {
let all_buckets = vec!["bucket-a".to_string()];
let set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let (info, _) = completed_data_usage_info_for_test(&[set], &all_buckets, false, false)
.expect("no-tier STANDARD/RRS usage does not require a tier map");
assert!(info.tier_stats.is_none());
}
#[test]
fn completed_data_usage_info_accepts_no_tier_unknown_and_standard_shape() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.size = 13;
entry.add_tier_sizes(&HashMap::from([(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: 3,
num_versions: 1,
num_objects: 1,
},
)]));
entry.add_unknown_tier_stats(&UnknownTierStats {
unknown_bytes: 9,
unknown_physical_bytes: 3,
unknown_objects: 1,
unknown_versions: 1,
..Default::default()
});
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 13,
logical_known: 4,
physical_total: 7,
physical_known: 4,
..Default::default()
});
let (info, _) = completed_data_usage_info_for_test(&[set], &all_buckets, false, false)
.expect("no-tier STANDARD plus UNKNOWN should remain publishable");
assert_eq!(
info.tier_stats.expect("unknown bucket should be retained").tiers[UNKNOWN_TIER].total_size,
3
);
}
#[test]
fn completed_data_usage_info_accepts_unknown_only_with_current_registry_generation() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
set.info.tier_registry_generation = Some(7);
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.size = 13;
entry.add_tier_sizes(&HashMap::from([(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: 3,
num_versions: 1,
num_objects: 1,
},
)]));
entry.add_unknown_tier_stats(&UnknownTierStats {
unknown_bytes: 9,
unknown_physical_bytes: 3,
unknown_objects: 1,
unknown_versions: 1,
..Default::default()
});
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 13,
logical_known: 4,
physical_total: 7,
physical_known: 4,
..Default::default()
});
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
assert!(
completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_some()
);
}
#[test]
fn completed_data_usage_info_rejects_non_registry_tier_in_current_generation() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
set.info.tier_registry_generation = Some(7);
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_tier_sizes(&HashMap::from([
(
"WARM".to_string(),
TierStats {
total_size: 4,
num_versions: 1,
num_objects: 1,
},
),
(
"RETIRED".to_string(),
TierStats {
total_size: 6,
num_versions: 1,
num_objects: 1,
},
),
]));
entry.tier_accounting_proof = Some(TierAccountingProof {
logical_total: 10,
logical_known: 10,
physical_total: 10,
physical_known: 10,
..Default::default()
});
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
assert!(
completed_data_usage_info(&[set], &expected_sources, &all_buckets, &["WARM".to_string()], true, false, false,).is_none()
);
}
#[test]
fn completed_data_usage_info_rejects_legacy_proof_missing_when_tier_accounted() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.add_tier_sizes(&HashMap::from([(
"WARM".to_string(),
TierStats {
total_size: 10,
num_versions: 1,
num_objects: 1,
},
)]));
entry.tier_accounting_proof = None;
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_rejects_legacy_proof_missing_for_scalar_usage() {
let all_buckets = vec!["bucket-a".to_string()];
let mut set = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let entry = set.cache.get_mut(&hash_path("bucket-a").key()).expect("bucket entry");
entry.tier_accounting_proof = None;
assert!(completed_data_usage_info_for_test(&[set], &all_buckets, false, false).is_none());
}
#[test]
fn completed_data_usage_info_omits_tier_stats_without_tiered_objects() {
let all_buckets = vec!["bucket-a".to_string()];
@@ -435,51 +156,6 @@ fn completed_data_usage_info_omits_tier_stats_without_tiered_objects() {
assert!(data_usage_info.tier_stats.is_none());
}
#[test]
fn completed_data_usage_info_rejects_legacy_and_new_tier_generations_mixed() {
let all_buckets = vec!["bucket-a".to_string()];
let legacy = completed_root_cache("bucket-a", 1, 10, DataUsageCacheSource::new(0, 0));
let mut current = completed_root_cache("bucket-a", 1, 20, DataUsageCacheSource::new(1, 0));
current.info.tier_registry_generation = Some(42);
assert!(
completed_data_usage_info_for_test(&[legacy, current], &all_buckets, false, false).is_none(),
"legacy and generation-tagged sets must not publish a mixed snapshot"
);
}
#[test]
fn current_cache_root_with_new_tier_generation_resets_old_cache() {
let source = DataUsageCacheSource::new(0, 0);
let mut cache = completed_root_cache("bucket-a", 1, 10, source);
cache.info.tier_registry_generation = Some(1);
let state = current_cache_root_or_prepare_with_generation(
&mut cache,
DATA_USAGE_ROOT,
source,
0,
0,
TEST_PLAN_DIGEST,
DataUsageCacheReuseOptions {
require_source: false,
tier_registry_generation: Some(2),
},
);
assert!(matches!(
state,
DataUsageCacheScanState::Prepared {
outcome: DataUsageCachePrepareOutcome::Reset,
..
}
));
assert!(cache.cache.is_empty(), "old-generation entries must not be reused");
assert_eq!(cache.info.tier_registry_generation, None);
assert_eq!(cache.info.scan_plan_digest, Some(TEST_PLAN_DIGEST));
assert!(!cache.info.snapshot_complete);
}
#[test]
fn completed_data_usage_info_requires_every_set_before_publish() {
let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string(), "bucket-empty".to_string()];
@@ -604,13 +280,6 @@ fn completed_data_usage_info_flattens_nested_bucket_entries() {
replica_size: 2048,
replica_count: 2,
}),
tier_accounting_proof: Some(TierAccountingProof {
logical_total: 2048,
logical_known: 2048,
physical_total: 2048,
physical_known: 2048,
..Default::default()
}),
..Default::default()
};
nested.obj_sizes.add(2048);
@@ -678,8 +347,7 @@ fn completed_data_usage_info_requires_exact_topology_sources() {
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0), DataUsageCacheSource::new(1, 0)]);
assert!(
completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, &[], true, false, false)
.is_none()
completed_data_usage_info(&[first_set, unexpected_set], &expected_sources, &all_buckets, true, false, false).is_none()
);
}
@@ -689,7 +357,7 @@ fn completed_data_usage_info_rejects_incomplete_bucket_plan() {
let set = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
let expected_sources = HashSet::from([DataUsageCacheSource::new(0, 0)]);
assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, &[], false, false, false).is_none());
assert!(completed_data_usage_info(&[set], &expected_sources, &all_buckets, false, false, false).is_none());
}
#[test]
+5 -44
View File
@@ -23,7 +23,7 @@ use crate::storage_api::owner::{
use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
use crate::{
DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
ScannerPutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
};
use rustfs_filemeta::FileInfo;
@@ -986,8 +986,8 @@ fn is_xl_meta_path_accepts_forward_separator() {
fn tier_stats_template_seeds_tiers_and_standard_classes() {
let template = tier_stats_template(&["WARM".to_string(), "COLD".to_string()]);
assert_eq!(template.len(), 5);
for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS, UNKNOWN_TIER] {
assert_eq!(template.len(), 4);
for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS] {
assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}");
}
}
@@ -1328,7 +1328,7 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() {
);
let update_time = SystemTime::now();
assert!(apply_bucket_result_to_cache(
apply_bucket_result_to_cache(
&mut cache,
DataUsageEntryInfo {
name: "bucket".to_string(),
@@ -1338,51 +1338,12 @@ fn apply_bucket_result_to_cache_updates_bucket_entry() {
objects: 2,
..Default::default()
},
tier_registry_generation: None,
},
update_time,
));
);
assert_eq!(cache.info.last_update, Some(update_time));
let entry = cache.find("bucket").expect("bucket entry should remain present");
assert_eq!(entry.size, 10);
assert_eq!(entry.objects, 2);
}
#[test]
fn apply_bucket_result_to_cache_rejects_a_different_tier_generation() {
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: DATA_USAGE_ROOT.to_string(),
tier_registry_generation: Some(7),
..Default::default()
},
..Default::default()
};
cache.replace(
"bucket",
DATA_USAGE_ROOT,
DataUsageEntry {
size: 3,
..Default::default()
},
);
let applied = apply_bucket_result_to_cache(
&mut cache,
DataUsageEntryInfo {
name: "bucket".to_string(),
parent: DATA_USAGE_ROOT.to_string(),
entry: DataUsageEntry {
size: 11,
..Default::default()
},
tier_registry_generation: Some(8),
},
SystemTime::now(),
);
assert!(!applied);
assert_eq!(cache.find("bucket").map(|entry| entry.size), Some(3));
assert!(cache.info.last_update.is_none());
}
-3
View File
@@ -46,7 +46,6 @@ pub const NS_SCANNER_SERVER_EPOCH_QUERY: &str = "ns_scanner_server_epoch";
pub const NS_SCANNER_SESSION_ID_QUERY: &str = "ns_scanner_session_id";
pub const NS_SCANNER_SESSION_SEQUENCE_QUERY: &str = "ns_scanner_session_sequence";
pub const NS_SCANNER_PROTOCOL_VERSION_QUERY: &str = "ns_scanner_protocol";
pub const NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY: &str = "ns_scanner_tier_registry_generation";
pub const NS_SCANNER_PROTOCOL_VERSION: u16 = 3;
pub const SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION: u32 = 0;
pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 5;
@@ -58,8 +57,6 @@ pub struct NsScannerCapabilityResponse {
pub version: u16,
pub server_epoch: uuid::Uuid,
pub proof: Vec<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub supports_tier_registry_generation: Option<bool>,
}
pub mod admin;
+3
View File
@@ -43,6 +43,9 @@ allow-git = [
# RustFS fork carrying presigned expiry and constant-time authentication fixes.
# owner: rustfs-maintainers review: 2026-10
"https://github.com/rustfs/s3s.git",
# MiMalloc fork pinned for hotpath allocation counting support.
# owner: houseme review: 2026-10
"https://github.com/xonatius/mimalloc_rust.git",
]
[bans]
-149
View File
@@ -1,149 +0,0 @@
# CI gate matrix
This file is the source of truth for which validation runs on each event, its
configured wall-clock budget, and whether it can block a merge. Test taxonomy,
naming, and nextest serialization rules remain in [README.md](README.md); e2e
membership and counts remain in
[e2e-suite-inventory.md](e2e-suite-inventory.md).
The distinction between **required** and **report-only** is load-bearing:
a failing job blocks a merge only when its exact check name is present in the
live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
does not make a job required by itself.
## Required merge checks
The live `main` ruleset (`6436880`) currently requires exactly these contexts:
| Required context | Producer | Validation |
|---|---|---|
| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
For pull requests limited to the paths excluded by the main CI workflow,
`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
`Test and Lint` under the same names. It runs the real quick checks and the
planning-document guard; it does not claim that Rust compilation or runtime
tests ran. Despite the workflow name, these paths also include selected deploy,
workflow, and lock files.
Verify the live rule rather than trusting this snapshot before changing merge
policy:
```bash
gh api repos/rustfs/rustfs/rulesets/6436880 \
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
```
The ruleset currently has `strict_required_status_checks_policy=false`.
`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
them, but `End-to-End Tests (full merge gate)` is not currently a required
context. Therefore the repository is prepared to test a merge-queue SHA, but
the workflow alone does not prove that every merge passed that lane.
## Pull request and merge matrix
Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
means the result is visible and actionable but is not in the live required
context list.
| Event | Validation | Budget | Merge status | Reproduction |
|---|---|---:|---|---|
| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
instead of adding a second ad-hoc selector. Before a profile runs,
`scripts/check_test_wiring.py` compares its exact membership to the committed
digest so a silent test drop fails closed.
## Scheduled and manual validation
Scheduled lanes are independent fault domains. They do not block a pull
request, but their workflow-local gate can fail the run and scheduled failures
are routed to the shared failure-issue action. The scheduled-validation
watchdog and freshness workflow separately detect incomplete runs and missing
schedules.
| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
|---|---|---:|---|---|
| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
runs are debugging evidence and intentionally do not open scheduled-failure
issues. A manual performance run may explicitly allow a known regression; that
override must not be treated as an ordinary passing baseline.
## Release validation
Release validation is post-merge and tag-driven; it does not substitute for a
pull-request gate.
| Event | Validation | Budget | Result |
|---|---|---:|---|
| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
are backfill/debug paths and do not prove the automatic `workflow_run` chain.
## Evidence requirements
A green check is useful only when it proves the intended behavior ran:
- Record the exact commit SHA and run URL.
- Separate product failure from runner prerequisites, service readiness, and
cancellation. Repair the precondition, then rerun the exact workload.
- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
server logs where the workflow provides them.
- For a bug fix or a new fault checker, provide sensitivity evidence: the old
behavior or an intentional mutation must fail the new oracle, and the fixed
behavior must pass it.
- Never promote a report-only lane to required from one green run. Require at
least 14 days and 30 representative pull requests with at least 99% complete
execution, then update the ruleset and this table together.
## Change checklist
Update this file in the same pull request when any of these change:
- workflow triggers, job names, timeouts, or nextest profile ownership;
- required status contexts or strict/merge-queue policy;
- scheduled cadence, alert routing, artifact contract, or local reproduction;
- report-only versus gating semantics.
Do not copy per-module test counts here. Update
[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
digest instead.
+2 -2
View File
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
rustfs-mimalloc = { workspace = true }
mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
rustfs-mimalloc-sys.workspace = true
libmimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+7 -1
View File
@@ -369,8 +369,14 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> {
rustfs_mimalloc::MiMalloc::collect(force);
// SAFETY: `mi_collect` is provided by the active global allocator backend
// on this target family. It is explicitly intended to reclaim retained
// pages/segments and does not require additional invariants from the caller.
unsafe {
libmimalloc_sys::mi_collect(force);
}
Ok(())
}
+8 -10
View File
@@ -26,22 +26,22 @@ struct MiMallocAllocator;
unsafe impl GlobalAlloc for MiMallocAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
unsafe { mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
}
}
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator]
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,9 +71,8 @@ mod tests {
allocation.extend_from_slice(&[7_u8; 64]);
assert_eq!(allocation.len(), 64);
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: the live Vec pointer is valid to inspect for heap ownership.
assert!(unsafe { heap.contains(allocation.as_ptr()) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
}
#[test]
@@ -86,13 +85,12 @@ mod tests {
let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
let allocator = super::MiMallocAllocator;
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: The pointer is checked for null before use and later released
// through the same allocator with the corresponding layout.
let ptr = unsafe { allocator.alloc_zeroed(layout) };
assert!(!ptr.is_null());
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -104,7 +102,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { heap.contains(grown_ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+36 -19
View File
@@ -17,7 +17,10 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split,
};
use serde::Serialize;
#[cfg(any(test, not(target_os = "windows")))]
use serde_json::Value;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
@@ -228,18 +231,7 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
let json = rustfs_mimalloc::MiMalloc::stats_json();
if json.is_empty() {
return None;
}
let observation = parse_mimalloc_stats_json(&json)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -250,6 +242,7 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
@@ -261,6 +254,7 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
@@ -277,10 +271,12 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current")
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
metrics
.iter()
@@ -289,6 +285,7 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
.filter(|value| *value > 0)
}
#[cfg(any(test, not(target_os = "windows")))]
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?;
let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -315,6 +312,33 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
let observation = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
libmimalloc_sys::mi_free(stats_ptr.cast());
observation?
};
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -542,13 +566,6 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[test]
fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
let snapshot = super::read_allocator_memory_snapshot();
#[cfg(not(target_os = "windows"))]
assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
}
#[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false);
+29 -84
View File
@@ -19,15 +19,13 @@ use crate::storage::storage_api::rpc_consumer::http_service::{
DEFAULT_READ_BUFFER_SIZE, DeleteOptions, DiskStore, NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse,
PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, StorageDiskRpcExt as _,
WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, check_and_record_signed_rpc_nonce, find_local_disk_by_ref,
sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, verify_put_file_auth_trailer,
verify_rpc_signature,
sign_ns_scanner_capability, sign_put_file_capability, verify_put_file_auth_trailer, verify_rpc_signature,
};
#[cfg(test)]
use crate::storage::storage_api::rpc_consumer::http_service::{
NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY,
NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY,
NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY, PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY,
WALK_DIR_BODY_SHA256_QUERY,
PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, WALK_DIR_BODY_SHA256_QUERY,
};
use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
use crate::storage::storage_api::tonic_rpc_auth_failure_reason;
@@ -292,8 +290,6 @@ struct NsScannerQuery {
struct NsScannerCapabilityQuery {
ns_scanner_protocol: Option<u16>,
ns_scanner_challenge: Option<uuid::Uuid>,
#[serde(rename = "ns_scanner_tier_registry_generation")]
ns_scanner_tier_registry_generation: Option<bool>,
}
fn verify_ns_scanner_body_digest(query: &NsScannerQuery, body: &[u8]) -> bool {
@@ -414,9 +410,7 @@ async fn handle_internode_rpc(req: Request<Incoming>) -> Response<Body> {
(Method::GET, WALK_DIR_PATH) | (Method::HEAD, WALK_DIR_PATH) => handle_walk_dir(req).await,
(Method::GET, NS_SCANNER_PATH) => match parse_query::<NsScannerCapabilityQuery>(&req) {
Ok(query) if query.ns_scanner_protocol == Some(NS_SCANNER_PROTOCOL_VERSION) => match query.ns_scanner_challenge {
Some(challenge) if !challenge.is_nil() => {
ns_scanner_capability_response(challenge, query.ns_scanner_tier_registry_generation == Some(true))
}
Some(challenge) if !challenge.is_nil() => ns_scanner_capability_response(challenge),
Some(_) | None => response_with_status(StatusCode::BAD_REQUEST, "namespace scanner challenge is invalid"),
},
Ok(_) => response_with_status(StatusCode::UPGRADE_REQUIRED, "namespace scanner protocol is unsupported"),
@@ -472,34 +466,31 @@ fn record_internode_rpc_error(operation: Option<&'static str>) {
}
}
fn ns_scanner_capability_response(challenge: uuid::Uuid, include_tier_registry_generation: bool) -> Response<Body> {
fn ns_scanner_capability_response(challenge: uuid::Uuid) -> Response<Body> {
let server_epoch = *NS_SCANNER_SERVER_EPOCH;
let proof =
match sign_ns_scanner_capability_with_tier_registry_generation(challenge, server_epoch, include_tier_registry_generation)
{
Ok(proof) => proof,
Err(err) => {
error!(
event = EVENT_RPC_REQUEST_FAILED,
component = LOG_COMPONENT_INTERNODE_RPC,
subsystem = LOG_SUBSYSTEM_NAMESPACE_SCANNER,
operation = INTERNODE_OPERATION_NS_SCANNER,
result = "failed",
status_code = StatusCode::UPGRADE_REQUIRED.as_u16(),
rpc_path = NS_SCANNER_PATH,
method = %Method::GET,
reason = "capability_authentication_unavailable",
error = %err,
"internode rpc request failed"
);
return response_with_status(StatusCode::UPGRADE_REQUIRED, "namespace scanner RPC authentication is unavailable");
}
};
let proof = match sign_ns_scanner_capability(challenge, server_epoch) {
Ok(proof) => proof,
Err(err) => {
error!(
event = EVENT_RPC_REQUEST_FAILED,
component = LOG_COMPONENT_INTERNODE_RPC,
subsystem = LOG_SUBSYSTEM_NAMESPACE_SCANNER,
operation = INTERNODE_OPERATION_NS_SCANNER,
result = "failed",
status_code = StatusCode::UPGRADE_REQUIRED.as_u16(),
rpc_path = NS_SCANNER_PATH,
method = %Method::GET,
reason = "capability_authentication_unavailable",
error = %err,
"internode rpc request failed"
);
return response_with_status(StatusCode::UPGRADE_REQUIRED, "namespace scanner RPC authentication is unavailable");
}
};
let body = match rmp_serde::to_vec_named(&NsScannerCapabilityResponse {
version: NS_SCANNER_PROTOCOL_VERSION,
server_epoch,
proof,
supports_tier_registry_generation: include_tier_registry_generation.then_some(true),
}) {
Ok(body) => body,
Err(err) => {
@@ -1685,13 +1676,12 @@ mod tests {
LOG_SUBSYSTEM_NAMESPACE_SCANNER, LOG_SUBSYSTEM_ROUTING, NS_SCANNER_BODY_SHA256_QUERY,
NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY, NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_PATH,
NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY, NS_SCANNER_SESSION_ID_QUERY,
NS_SCANNER_SESSION_SEQUENCE_QUERY, NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY, NsScannerCapabilityResponse,
NsScannerQuery, PUT_FILE_AUTH_STREAM_PATH, PUT_FILE_CAPABILITY_PATH, PUT_FILE_STREAM_PATH, PutFileQuery,
READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, append_walk_dir_completion,
internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, ns_scanner_response_body,
ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce, put_file_capability_response,
put_file_server_epoch_matches, put_file_stage_error_message, put_file_target_lock, read_file_body_stream,
read_file_stream_buffer_size, remote_scanner_claim_rejection, response_with_disk_error,
NS_SCANNER_SESSION_SEQUENCE_QUERY, NsScannerQuery, PUT_FILE_AUTH_STREAM_PATH, PUT_FILE_CAPABILITY_PATH,
PUT_FILE_STREAM_PATH, PutFileQuery, READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery,
append_walk_dir_completion, internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path,
ns_scanner_response_body, ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce,
put_file_capability_response, put_file_server_epoch_matches, put_file_stage_error_message, put_file_target_lock,
read_file_body_stream, read_file_stream_buffer_size, remote_scanner_claim_rejection, response_with_disk_error,
supports_walk_dir_stream_completion, validate_walk_dir_completion_request, verify_internode_rpc_signature,
verify_ns_scanner_body_digest, verify_walk_dir_body_digest, walk_dir_response_body, write_authenticated_put_file,
write_body_chunks_to_writer, write_put_file_body_chunks_to_writer,
@@ -2124,51 +2114,6 @@ mod tests {
);
assert!(serde_urlencoded::from_str::<NsScannerQuery>(&query).is_err());
assert!(serde_urlencoded::from_str::<super::NsScannerCapabilityQuery>("ns_scanner_protocol=1&unexpected=true").is_err());
let marked =
format!("ns_scanner_protocol=3&ns_scanner_challenge={request_id}&{NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY}=true");
assert_eq!(
serde_urlencoded::from_str::<super::NsScannerCapabilityQuery>(&marked)
.expect("generation marker should be accepted")
.ns_scanner_tier_registry_generation,
Some(true)
);
}
#[test]
fn namespace_scanner_capability_response_support_is_optional_for_old_peers() {
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyCapabilityResponse {
version: u16,
server_epoch: uuid::Uuid,
proof: Vec<u8>,
}
let old = rmp_serde::to_vec_named(&NsScannerCapabilityResponse {
version: super::NS_SCANNER_PROTOCOL_VERSION,
server_epoch: uuid::Uuid::new_v4(),
proof: vec![1, 2, 3],
supports_tier_registry_generation: None,
})
.expect("old response shape should encode");
let decoded: NsScannerCapabilityResponse = rmp_serde::from_slice(&old).expect("old response should decode");
assert_eq!(decoded.supports_tier_registry_generation, None);
let legacy_decoded: LegacyCapabilityResponse =
rmp_serde::from_slice(&old).expect("legacy reader should decode old shape");
assert_eq!(legacy_decoded.version, super::NS_SCANNER_PROTOCOL_VERSION);
assert!(!legacy_decoded.server_epoch.is_nil());
assert_eq!(legacy_decoded.proof, vec![1, 2, 3]);
let current = rmp_serde::to_vec_named(&NsScannerCapabilityResponse {
version: super::NS_SCANNER_PROTOCOL_VERSION,
server_epoch: uuid::Uuid::new_v4(),
proof: vec![1, 2, 3],
supports_tier_registry_generation: Some(true),
})
.expect("new response shape should encode");
let decoded: NsScannerCapabilityResponse = rmp_serde::from_slice(&current).expect("new response should decode");
assert_eq!(decoded.supports_tier_registry_generation, Some(true));
assert!(rmp_serde::from_slice::<LegacyCapabilityResponse>(&current).is_err());
}
#[test]
+9 -18
View File
@@ -224,8 +224,8 @@ pub(crate) mod rpc_consumer {
pub(crate) use super::super::storage_contracts::{
NS_SCANNER_BODY_SHA256_QUERY, NS_SCANNER_CAPABILITY_CHALLENGE_QUERY, NS_SCANNER_CYCLE_QUERY,
NS_SCANNER_LEADER_EPOCH_QUERY, NS_SCANNER_REQUEST_ID_QUERY, NS_SCANNER_SERVER_EPOCH_QUERY,
NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY,
PUT_FILE_CAPABILITY_CHALLENGE_QUERY, PUT_FILE_CAPABILITY_QUERY, WALK_DIR_BODY_SHA256_QUERY,
NS_SCANNER_SESSION_ID_QUERY, NS_SCANNER_SESSION_SEQUENCE_QUERY, PUT_FILE_CAPABILITY_CHALLENGE_QUERY,
PUT_FILE_CAPABILITY_QUERY, WALK_DIR_BODY_SHA256_QUERY,
};
pub(crate) use super::super::storage_contracts::{
NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse, PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1,
@@ -233,8 +233,8 @@ pub(crate) mod rpc_consumer {
};
pub(crate) use super::super::{
DeleteOptions, DiskStore, StorageDiskRpcExt, WalkDirOptions, check_and_record_signed_rpc_nonce,
find_local_disk_by_ref, sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability,
verify_put_file_auth_trailer, verify_rpc_signature,
find_local_disk_by_ref, sign_ns_scanner_capability, sign_put_file_capability, verify_put_file_auth_trailer,
verify_rpc_signature,
};
}
@@ -520,10 +520,9 @@ pub(crate) mod ecstore_rpc {
pub(crate) use rustfs_ecstore::api::rpc::{
KMS_SIGNAL_SUBSYSTEM, LocalPeerS3Client, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient,
PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX,
check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience,
sign_ns_scanner_capability_with_tier_registry_generation, sign_put_file_capability, sign_tonic_rpc_response_proof,
tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers, tonic_rpc_auth_failure_reason,
verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
check_and_record_signed_rpc_nonce, normalize_tonic_rpc_audience, sign_ns_scanner_capability, sign_put_file_capability,
sign_tonic_rpc_response_proof, tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers,
tonic_rpc_auth_failure_reason, verify_put_file_auth_trailer, verify_rpc_signature, verify_tonic_canonical_body_digest,
verify_tonic_mutation_body_digest, verify_tonic_rpc_signature_with_bootstrap,
};
#[cfg(test)]
@@ -1732,16 +1731,8 @@ pub(crate) fn verify_put_file_auth_trailer(
ecstore_rpc::verify_put_file_auth_trailer(url, method, nonce, trailer)
}
pub(crate) fn sign_ns_scanner_capability_with_tier_registry_generation(
challenge: uuid::Uuid,
server_epoch: uuid::Uuid,
supports_tier_registry_generation: bool,
) -> std::io::Result<Vec<u8>> {
ecstore_rpc::sign_ns_scanner_capability_with_tier_registry_generation(
challenge,
server_epoch,
supports_tier_registry_generation,
)
pub(crate) fn sign_ns_scanner_capability(challenge: uuid::Uuid, server_epoch: uuid::Uuid) -> std::io::Result<Vec<u8>> {
ecstore_rpc::sign_ns_scanner_capability(challenge, server_epoch)
}
pub(crate) fn sign_put_file_capability(
+1 -1
View File
@@ -241,7 +241,7 @@ env \
RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
cargo test -p rustfs-kms --test vault_ha_failover_live \
vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
--ignored --nocapture --test-threads=1 &
TEST_PID=$!