Compare commits

...

13 Commits

Author SHA1 Message Date
houseme 16fc3ff0e7 Merge branch 'main' into cxymds/fix-1935-size-resolution-unknown-run2
Signed-off-by: houseme <housemecn@gmail.com>
2026-08-23 12:32:56 +08:00
Zhengchao An 23a2c7d776 test(kms): stabilize Vault failover validation (#6385)
* test(kms): bound Vault failover progress wait

* ci(nightly): honor manual dispatch ref

* test(kms): preserve Vault worker failures

* test(kms): validate Vault circuit recovery
2026-08-23 12:32:11 +08:00
唐小鸭 5f72209446 fix(ecstore): keep unknown-size sentinel in create_bitrot_writer (#6380)
SSE and compression wrap the payload so its length is unknown and
advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved
that sentinel except create_bitrot_writer, which clamped it to 0 before
calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in
the put_file_stream query, so remote peers were told the body was empty.

Since the authenticated put-file trailer (#5868) the receiver used the
declared size to split body from trailer, turning the clamp into a fatal
"auth trailer has trailing data" failure for every SSE PUT on multi-node
deployments (rc.2). #6320 relaxed the receiver to only trust size > 0;
this change fixes the sender so the sentinel survives end to end and the
wire no longer conflates empty objects with unknown-length streams.

Refs #6331
2026-08-23 12:29:52 +08:00
Zhengchao An b6ba89d9e4 docs(testing): document CI gate matrix (#6412) 2026-08-23 12:09:06 +08:00
houseme 648d5166e2 feat(allocator): replace mimalloc/libmimalloc-sys with rustfs-mimalloc/rustfs-mimalloc-sys (#6404)
Replace the upstream xonatius/mimalloc_rust.git fork (mimalloc + libmimalloc-sys)
with the published rustfs-mimalloc (v0.5.0) and rustfs-mimalloc-sys (v0.5.0) crates
from crates.io.

The new crates are based on mimalloc V3 (v3.5.0) and provide:
- MiMalloc global allocator with safe API (collect, stats_json, process_info)
- Heap management and arena operations (heap module)
- Full FFI bindings to mimalloc V3

Changes:
- Workspace deps: mimalloc + libmimalloc-sys (git) → rustfs-mimalloc + rustfs-mimalloc-sys (crates.io)
- allocator_reclaim.rs: libmimalloc_sys::mi_collect → rustfs_mimalloc::MiMalloc::collect
- memory_observability.rs: raw FFI mi_stats_get_json → MiMalloc::stats_json()
- main.rs: heap ownership tests use Heap::contains() (V3 API)
- deny.toml: remove xonatius/mimalloc_rust.git from allow-git

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-23 12:07:25 +08:00
houseme 84eb5aebef fix(ecstore): remove inline write debug noise (#6408)
* fix(ecstore): remove inline write debug noise

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

* fix(ecstore): satisfy warning-as-error lints

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-23 12:07:20 +08:00
马登山 6dbff1c036 fix(scanner): close lifecycle review gaps 2026-08-23 09:45:36 +08:00
马登山 854b6dfb0c fix(ci): resolve ecstore clippy warnings 2026-08-23 09:20:07 +08:00
马登山 893a4a11c1 Merge remote-tracking branch 'origin/main' into HEAD 2026-08-23 09:17:13 +08:00
马登山 2086ade967 fix(scanner): preserve restore expiry semantics 2026-08-23 08:41:31 +08:00
cxymds 20d1266496 fix(heal): fence format repair during pool transitions (#6342)
* fix(heal): fence format repair during pool transitions

* fix(heal): fence format writes during transitions
2026-08-23 04:24:48 +08:00
唐小鸭 f7003dfddd fix(admin): four site-replication interop correctness fixes (B5-rc T2) (#6399)
* fix(admin): send versioningEnabled on site replication make-bucket ops

The outbound make-with-versioning bucket-op query only carried
operation/createdAt/lockEnabled. MinIO's own create-bucket hook sends
versioningEnabled=true on this op, so align the outbound query with
MinIO's site-replication make-bucket wire contract. Route both outbound
builders (bootstrap plan and create-bucket hook) through one shared
builder that always appends versioningEnabled=true. RustFS's own inbound
handler force-enables versioning either way, so RustFS-to-RustFS
behavior is unchanged; the MinIO release verified against
(RELEASE.2025-09-07) also force-enables versioning regardless of the
flag, so this aligns the wire contract rather than changing observable
behavior there.

* fix(admin): propagate purge-deleted-bucket errors in site replication

The purge-deleted-bucket branch of the peer bucket-ops handler dropped
the delete_bucket error and answered 200, so a peer-driven purge that
failed (disk full, quorum loss) was reported as success while the
bucket survived on this site. Tolerate only bucket-not-found (the purge
raced an earlier replay or a local delete) and propagate every other
error through ApiError like the sibling delete branches do.

* fix(admin): derive fallback site deployment ID with UUIDv5

deployment_id_for_endpoint used DefaultHasher, whose algorithm is not
guaranteed stable across Rust releases. The fallback fires when a peer
response carries an empty deploymentID; the result is persisted in
site-replication state, used for collision disambiguation, and
broadcast to peers, so a toolchain bump could re-derive a different ID
for the same endpoint. Note that the add preflight currently rejects
that case upstream of this fallback. Derive UUIDv5 (NAMESPACE_URL) over
the canonical endpoint instead, and log a structured warn when a peer
metainfo response arrives without a deploymentID. Already persisted
fallback IDs are non-empty and therefore never re-derived, so existing
state is unaffected.

* fix(admin): stream site replication devnull body without 1MB cap

The site-replication devnull endpoint buffered the request body through
read_plain_admin_body, which enforces the 1MB admin body cap. MinIO
peers stream multi-megabyte probe bodies to this endpoint during site
netperf link checks and expect an unbounded discard, so any larger
probe got a 400 and was misreported as a broken link. Stream and
discard the body chunk by chunk with no size cap instead, mirroring
MinIO's io.Discard drain. The response stays 204 with an empty body.
2026-08-23 04:24:04 +08:00
马登山 23b85b792e fix(scanner): unify unknown metadata size accounting 2026-08-22 22:58:05 +08:00
33 changed files with 2104 additions and 274 deletions
+3 -6
View File
@@ -39,11 +39,10 @@ jobs:
env: env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps: steps:
- name: Checkout main branch - name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with: with:
persist-credentials: false persist-credentials: false
ref: main
- name: Setup Rust environment - name: Setup Rust environment
uses: ./.github/actions/setup uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
# either casing. # either casing.
NO_PROXY: 127.0.0.1,localhost NO_PROXY: 127.0.0.1,localhost
steps: steps:
- name: Checkout main branch - name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with: with:
persist-credentials: false persist-credentials: false
ref: main
- name: Setup Rust environment - name: Setup Rust environment
uses: ./.github/actions/setup uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
NO_PROXY: 127.0.0.1,localhost NO_PROXY: 127.0.0.1,localhost
steps: steps:
- name: Checkout main branch - name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with: with:
persist-credentials: false persist-credentials: false
ref: main
- name: Setup Rust environment - name: Setup Rust environment
uses: ./.github/actions/setup uses: ./.github/actions/setup
+2 -1
View File
@@ -30,7 +30,8 @@ make build-docker BUILD_OS=ubuntu22.04
- Crate membership: `Cargo.toml` `[workspace].members` - Crate membership: `Cargo.toml` `[workspace].members`
- Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md) - Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
- Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md) - Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs) - CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake - Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md) policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing): - Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
+2
View File
@@ -70,6 +70,8 @@ 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 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 ### 🔒 Automated Pre-commit Hooks
#### What `make pre-commit` and `make pre-pr` actually run #### What `make pre-commit` and `make pre-pr` actually run
Generated
+23 -27
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]] [[package]]
name = "cc" name = "cc"
version = "1.4.3" version = "1.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
dependencies = [ dependencies = [
"find-msvc-tools", "find-msvc-tools",
"jobserver", "jobserver",
@@ -2522,12 +2522,6 @@ dependencies = [
"subtle", "subtle",
] ]
[[package]]
name = "cty"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
[[package]] [[package]]
name = "curve25519-dalek" name = "curve25519-dalek"
version = "4.1.3" version = "4.1.3"
@@ -5988,15 +5982,6 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" 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]] [[package]]
name = "libredox" name = "libredox"
version = "0.1.20" version = "0.1.20"
@@ -6397,14 +6382,6 @@ dependencies = [
"synstructure 0.13.2", "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]] [[package]]
name = "mime" name = "mime"
version = "0.3.17" version = "0.3.17"
@@ -9162,13 +9139,11 @@ dependencies = [
"insta", "insta",
"jiff", "jiff",
"libc", "libc",
"libmimalloc-sys",
"libsystemd", "libsystemd",
"matchit 0.9.2", "matchit 0.9.2",
"md-5 0.11.0", "md-5 0.11.0",
"metrics", "metrics",
"metrics-util", "metrics-util",
"mimalloc",
"mime_guess", "mime_guess",
"opentelemetry", "opentelemetry",
"opentelemetry_sdk", "opentelemetry_sdk",
@@ -9204,6 +9179,8 @@ dependencies = [
"rustfs-lock", "rustfs-lock",
"rustfs-log-analyzer", "rustfs-log-analyzer",
"rustfs-madmin", "rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify", "rustfs-notify",
"rustfs-object-capacity", "rustfs-object-capacity",
"rustfs-object-data-cache", "rustfs-object-data-cache",
@@ -9875,6 +9852,24 @@ dependencies = [
"tokio", "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]] [[package]]
name = "rustfs-notify" name = "rustfs-notify"
version = "1.0.0-rc.3" version = "1.0.0-rc.3"
@@ -12688,6 +12683,7 @@ dependencies = [
"js-sys", "js-sys",
"rand 0.10.2", "rand 0.10.2",
"serde_core", "serde_core",
"sha1_smol",
"wasm-bindgen", "wasm-bindgen",
] ]
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0" dav-server = "0.11.0"
# Performance Analysis and Memory Profiling # Performance Analysis and Memory Profiling
mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" } rustfs-mimalloc = { version = "0.5.0" }
libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] } rustfs-mimalloc-sys = { version = "0.5.0" }
hotpath = { version = "0.23.3", default-features = false } hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection # Snapshot testing for output format regression detection
insta = { version = "1.48" } insta = { version = "1.48" }
+89 -1
View File
@@ -307,6 +307,38 @@ pub struct DiskUsageStatus {
pub snapshot_exists: bool, pub snapshot_exists: bool,
} }
/// A bounded reconciliation record for an object whose logical size could not
/// be trusted at the scanner boundary. The scanner persists these records in
/// its cache; keeping the model here avoids a second, incompatible accounting
/// representation in storage-facing crates.
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SizeReconciliationEntry {
/// Stable object/version identity key (not a metrics label).
pub key: String,
pub bucket: String,
pub object: String,
#[serde(default)]
pub version_id: Option<String>,
#[serde(default)]
pub generation: Option<String>,
/// Structured reason label; raw metadata values must never be stored here.
pub reason: String,
#[serde(default)]
pub physical_size: Option<u64>,
#[serde(default)]
pub first_seen: u64,
#[serde(default)]
pub attempts: u32,
}
/// Object scope refreshed by one scanner pass. Existing debts in this scope
/// are removed before the pass's unresolved records are inserted.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct SizeReconciliationScope {
pub bucket: String,
pub object: String,
}
/// Size summary for a single object or group of objects /// Size summary for a single object or group of objects
#[derive(Debug, Default, Clone)] #[derive(Debug, Default, Clone)]
pub struct SizeSummary { pub struct SizeSummary {
@@ -336,6 +368,16 @@ pub struct SizeSummary {
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>, pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
/// Per-tier accounting, keyed by storage class or remote tier name /// Per-tier accounting, keyed by storage class or remote tier name
pub tier_stats: HashMap<String, TierStats>, pub tier_stats: HashMap<String, TierStats>,
/// Size-resolution debts observed while scanning this summary.
pub size_reconciliation: Vec<SizeReconciliationEntry>,
/// True when the per-object summary exceeded its bounded debt buffer.
/// Callers must retain prior ledger entries rather than treating the
/// partial list as a complete refresh.
pub size_reconciliation_truncated: bool,
/// Object scopes refreshed by this summary. They let the durable ledger
/// remove versions that resolved without allocating one key per healthy
/// version on the hot path.
pub reconciliation_scopes: Vec<SizeReconciliationScope>,
} }
/// Replication target size summary /// Replication target size summary
@@ -833,7 +875,8 @@ impl DataUsageEntry {
/// ///
/// The canonical wire format is written by the hand-written map-encoded /// The canonical wire format is written by the hand-written map-encoded
/// `Serialize` on the scanner-side `DataUsageCacheInfo` /// `Serialize` on the scanner-side `DataUsageCacheInfo`
/// (`crates/scanner/src/data_usage_define.rs`), which carries 16 fields. /// (`crates/scanner/src/data_usage_define.rs`), which carries the original 16
/// fields plus an optional reconciliation field.
/// This type decodes only the shared subset and is deliberately not /// This type decodes only the shared subset and is deliberately not
/// `Serialize`: a derived (array) encoding of this 6-field subset would /// `Serialize`: a derived (array) encoding of this 6-field subset would
/// corrupt the cache for scanner readers, so no write path may exist here. /// corrupt the cache for scanner readers, so no write path may exist here.
@@ -1777,6 +1820,51 @@ impl SizeSummary {
entry.pending_count = entry.pending_count.saturating_add(stats.pending_count); entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
entry.failed_count = entry.failed_count.saturating_add(stats.failed_count); entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
} }
for entry in &other.size_reconciliation {
self.record_size_reconciliation(entry.clone());
}
self.size_reconciliation_truncated |= other.size_reconciliation_truncated;
for scope in &other.reconciliation_scopes {
self.record_reconciliation_scope(&scope.bucket, &scope.object);
}
}
/// Add one reconciliation debt, coalescing repeated observations in the
/// same object summary. The scanner cache applies its own larger bound.
pub fn record_size_reconciliation(&mut self, entry: SizeReconciliationEntry) {
const MAX_SUMMARY_RECONCILIATION_ENTRIES: usize = 1024;
if let Some(existing) = self.size_reconciliation.iter_mut().find(|value| value.key == entry.key) {
existing.reason = entry.reason;
existing.physical_size = entry.physical_size;
existing.generation = entry.generation;
existing.version_id = entry.version_id;
return;
}
if self.size_reconciliation.len() < MAX_SUMMARY_RECONCILIATION_ENTRIES {
self.size_reconciliation.push(entry);
} else {
self.size_reconciliation_truncated = true;
}
}
/// Mark one object scope as refreshed. Duplicate scopes are suppressed so
/// merging summaries remains bounded and deterministic.
pub fn record_reconciliation_scope(&mut self, bucket: &str, object: &str) {
if !self
.reconciliation_scopes
.iter()
.any(|scope| scope.bucket == bucket && scope.object == object)
{
if self.reconciliation_scopes.len() >= 1024 {
self.size_reconciliation_truncated = true;
return;
}
self.reconciliation_scopes.push(SizeReconciliationScope {
bucket: bucket.to_string(),
object: object.to_string(),
});
}
} }
} }
+1 -1
View File
@@ -2026,7 +2026,7 @@ impl PoolMeta {
self.load_no_lock(pool).await self.load_no_lock(pool).await
} }
async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()> pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
where where
S: EcstoreObjectIO, S: EcstoreObjectIO,
{ {
+20 -8
View File
@@ -988,14 +988,11 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for Sets {
} }
} }
#[async_trait::async_trait] impl Sets {
impl crate::storage_api_contracts::heal::HealOperations for Sets { pub(crate) async fn heal_format_with_fence<F>(&self, dry_run: bool, fence_lost: F) -> Result<(HealResultItem, Option<Error>)>
type Error = Error; where
type HealResultItem = HealResultItem; F: Fn() -> bool + Send + Sync,
type HealOptions = HealOpts; {
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
let (disks, init_errs) = init_storage_disks_with_errors( let (disks, init_errs) = init_storage_disks_with_errors(
&self.endpoints.endpoints, &self.endpoints.endpoints,
&DiskOption { &DiskOption {
@@ -1068,6 +1065,9 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
// Save new formats `format.json` on unformatted disks. // Save new formats `format.json` on unformatted disks.
for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() { for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() {
if fm.is_some() && disk.is_some() { if fm.is_some() && disk.is_some() {
if fence_lost() {
return Ok((res, Some(StorageError::SlowDown)));
}
if let Err(err) = save_format_file(disk, fm).await { if let Err(err) = save_format_file(disk, fm).await {
if let Some(disk) = disk.as_ref() { if let Some(disk) = disk.as_ref() {
let _ = disk.close().await; let _ = disk.close().await;
@@ -1101,6 +1101,18 @@ impl crate::storage_api_contracts::heal::HealOperations for Sets {
} }
Ok((res, None)) Ok((res, None))
} }
}
#[async_trait::async_trait]
impl crate::storage_api_contracts::heal::HealOperations for Sets {
type Error = Error;
type HealResultItem = HealResultItem;
type HealOptions = HealOpts;
#[tracing::instrument(skip(self))]
async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
self.heal_format_with_fence(dry_run, || false).await
}
#[tracing::instrument(skip(self))] #[tracing::instrument(skip(self))]
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> { async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let mut result = HealResultItem { let mut result = HealResultItem {
+38 -6
View File
@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
/// ///
/// # Returns /// # Returns
/// A Result containing the BitrotWriterWrapper or an error /// 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( pub async fn create_bitrot_writer(
is_inline_buffer: bool, is_inline_buffer: bool,
disk: Option<&DiskStore>, disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
let writer = if is_inline_buffer { let writer = if is_inline_buffer {
CustomWriter::new_inline_buffer() CustomWriter::new_inline_buffer()
} else if let Some(disk) = disk { } else if let Some(disk) = disk {
let length = if length > 0 { let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
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?; let file = disk.create_file("", volume, path, length).await?;
#[cfg(feature = "hotpath")] #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
use rustfs_rio::ChunkReader; use rustfs_rio::ChunkReader;
use std::collections::VecDeque; 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 { struct TestChunkReader {
chunks: VecDeque<Bytes>, chunks: VecDeque<Bytes>,
} }
+1 -14
View File
@@ -2124,26 +2124,13 @@ impl SetDisks {
let put_object_size = known_put_object_storage_size(data.size()); let put_object_size = known_put_object_storage_size(data.size());
let shard_file_size_raw = erasure.shard_file_size(put_object_size); let shard_file_size_raw = erasure.shard_file_size(put_object_size);
let is_inline_buffer = let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
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 collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
let shard_file_size = shard_file_size_raw; let shard_file_size = shard_file_size_raw;
let shard_size = erasure.shard_size(); let shard_size = erasure.shard_size();
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_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); 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()); rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let writer_setup_stage_start = collect_stage_timing.then(Instant::now); let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
let (mut writers, errors) = if direct_inline_commit { let (mut writers, errors) = if direct_inline_commit {
+332 -3
View File
@@ -13,7 +13,12 @@
// limitations under the License. // limitations under the License.
use super::*; use super::*;
use crate::core::pools::POOL_META_NAME;
use crate::services::rebalance::{REBAL_META_NAME, RebalStatus};
use crate::set_disk::get_lock_acquire_timeout;
use crate::storage_api_contracts::heal::HealOperations as _; use crate::storage_api_contracts::heal::HealOperations as _;
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use rustfs_lock::NamespaceLockGuard;
use tracing::trace; use tracing::trace;
const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -30,7 +35,119 @@ fn invalid_heal_pool_index(pool_idx: usize, pool_count: usize) -> Error {
) )
} }
#[derive(Debug, Clone, Copy)]
enum HealFormatPoolSkip {
Completed,
Retryable,
}
fn classify_heal_format_pool(
pool_idx: usize,
pool_cmd_line: &str,
pool_meta: &PoolMeta,
rebalance_meta: Option<&RebalanceMeta>,
) -> Option<HealFormatPoolSkip> {
let Some(pool) = pool_meta.pools.get(pool_idx) else {
return Some(HealFormatPoolSkip::Retryable);
};
if pool.id != pool_idx || pool_cmd_line.is_empty() || pool.cmd_line.is_empty() || pool.cmd_line != pool_cmd_line {
return Some(HealFormatPoolSkip::Retryable);
}
if let Some(decommission) = pool.decommission.as_ref() {
if decommission.complete {
return Some(HealFormatPoolSkip::Completed);
}
if decommission.failed || decommission.canceled || decommission.queued || pool_meta.is_suspended(pool_idx) {
return Some(HealFormatPoolSkip::Retryable);
}
}
if let Some(meta) = rebalance_meta {
let Some(pool_stats) = meta.pool_stats.get(pool_idx) else {
return Some(HealFormatPoolSkip::Retryable);
};
if pool_stats.info.stopping || (pool_stats.participating && pool_stats.info.status == RebalStatus::Started) {
return Some(HealFormatPoolSkip::Retryable);
}
}
None
}
fn heal_format_pool_skip_error(skip: HealFormatPoolSkip) -> Error {
match skip {
HealFormatPoolSkip::Completed => StorageError::NoHealRequired,
HealFormatPoolSkip::Retryable => StorageError::SlowDown,
}
}
fn heal_format_fence_lost_error() -> Error {
StorageError::SlowDown
}
impl ECStore { impl ECStore {
async fn acquire_heal_format_fence(
&self,
) -> Result<(NamespaceLockGuard, NamespaceLockGuard, PoolMeta, Option<RebalanceMeta>)> {
let metadata_pool = self
.pools
.first()
.cloned()
.ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
// Metadata fence order is part of the decommission/rebalance protocol:
// pool.bin must always be acquired before rebalance.bin.
let pool_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
let pool_guard = pool_lock.get_write_lock(get_lock_acquire_timeout()).await?;
let rebalance_lock = metadata_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
let rebalance_guard = rebalance_lock.get_write_lock(get_lock_acquire_timeout()).await?;
if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
return Err(heal_format_fence_lost_error());
}
let mut pool_meta = PoolMeta::default();
pool_meta.load_no_lock(metadata_pool.clone()).await?;
if pool_meta.pools.len() != self.pools.len()
|| pool_meta.pools.iter().enumerate().any(|(pool_idx, pool)| {
pool.id != pool_idx || pool.cmd_line.is_empty() || pool.cmd_line != self.pools[pool_idx].endpoints.cmd_line
})
{
return Err(heal_format_fence_lost_error());
}
let mut rebalance_meta = RebalanceMeta::new();
let rebalance_meta = match rebalance_meta
.load_with_opts(
metadata_pool,
ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
{
Ok(()) => Some(rebalance_meta),
Err(Error::ConfigNotFound) => None,
Err(err) => return Err(err),
};
if rebalance_meta
.as_ref()
.is_some_and(|meta| meta.pool_stats.len() != self.pools.len())
{
return Err(heal_format_fence_lost_error());
}
if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
return Err(heal_format_fence_lost_error());
}
Ok((pool_guard, rebalance_guard, pool_meta, rebalance_meta))
}
fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> { fn get_pools_for_heal_object(&self, opts: &HealOpts) -> Result<Vec<Arc<Sets>>> {
match opts.pool { match opts.pool {
Some(pool_idx) => Ok(vec![ Some(pool_idx) => Ok(vec![
@@ -52,9 +169,26 @@ impl ECStore {
}; };
let mut count_no_heal = 0; let mut count_no_heal = 0;
let mut count_completed = 0;
let mut first_error = None; let mut first_error = None;
for pool in self.pools.iter() { for (pool_idx, pool) in self.pools.iter().enumerate() {
let (mut result, err) = pool.heal_format(dry_run).await?; let (pool_guard, rebalance_guard, pool_meta, rebalance_meta) = self.acquire_heal_format_fence().await?;
if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
first_error.get_or_insert(heal_format_fence_lost_error());
break;
}
if let Some(skip) = classify_heal_format_pool(pool_idx, &pool.endpoints.cmd_line, &pool_meta, rebalance_meta.as_ref())
{
if matches!(skip, HealFormatPoolSkip::Completed) {
count_completed += 1;
} else {
first_error.get_or_insert(heal_format_pool_skip_error(skip));
}
continue;
}
let fence_lost = || pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost();
let (mut result, err) = pool.heal_format_with_fence(dry_run, fence_lost).await?;
if let Some(err) = err { if let Some(err) = err {
match err { match err {
StorageError::NoHealRequired => { StorageError::NoHealRequired => {
@@ -69,11 +203,18 @@ impl ECStore {
r.set_count += result.set_count; r.set_count += result.set_count;
r.before.drives.append(&mut result.before.drives); r.before.drives.append(&mut result.before.drives);
r.after.drives.append(&mut result.after.drives); r.after.drives.append(&mut result.after.drives);
// A lease can be lost after the final write; fail closed before
// reporting the pool as successfully healed.
if pool_guard.is_lock_lost() || rebalance_guard.is_lock_lost() {
first_error.get_or_insert(heal_format_fence_lost_error());
break;
}
} }
if let Some(err) = first_error { if let Some(err) = first_error {
return Ok((r, Some(err))); return Ok((r, Some(err)));
} }
if count_no_heal == self.pools.len() { if count_no_heal + count_completed == self.pools.len() {
info!( info!(
event = EVENT_HEAL_FORMAT_COMPLETED, event = EVENT_HEAL_FORMAT_COMPLETED,
component = LOG_COMPONENT_ECSTORE, component = LOG_COMPONENT_ECSTORE,
@@ -302,6 +443,7 @@ mod tests {
use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk}; use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use crate::runtime::instance::InstanceContext; use crate::runtime::instance::InstanceContext;
use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions}; use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations}; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
use crate::store::init_format::{load_format_erasure, save_format_file}; use crate::store::init_format::{load_format_erasure, save_format_file};
@@ -353,6 +495,164 @@ mod tests {
} }
} }
fn pool_meta_with_decommission(info: PoolDecommissionInfo) -> PoolMeta {
PoolMeta {
pools: vec![PoolStatus {
id: 0,
cmd_line: "pool-0".to_string(),
last_update: OffsetDateTime::UNIX_EPOCH,
decommission: Some(info),
}],
..Default::default()
}
}
#[test]
fn heal_format_pool_state_barriers_are_classified() {
let active = pool_meta_with_decommission(PoolDecommissionInfo {
start_time: Some(OffsetDateTime::UNIX_EPOCH),
..Default::default()
});
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &active, None),
Some(HealFormatPoolSkip::Retryable)
));
for info in [
PoolDecommissionInfo {
failed: true,
..Default::default()
},
PoolDecommissionInfo {
canceled: true,
..Default::default()
},
] {
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &pool_meta_with_decommission(info), None),
Some(HealFormatPoolSkip::Retryable)
));
}
let completed = pool_meta_with_decommission(PoolDecommissionInfo {
complete: true,
..Default::default()
});
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &completed, None),
Some(HealFormatPoolSkip::Completed)
));
}
#[test]
fn heal_format_pool_rebalance_barriers_and_identity_are_fail_closed() {
let identity_meta = pool_meta_with_decommission(PoolDecommissionInfo::default());
let rebalance = RebalanceMeta {
pool_stats: vec![RebalanceStats {
participating: true,
info: RebalanceInfo {
status: RebalStatus::Started,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&rebalance)),
Some(HealFormatPoolSkip::Retryable)
));
let stopping = RebalanceMeta {
pool_stats: vec![RebalanceStats {
info: RebalanceInfo {
stopping: true,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping)),
Some(HealFormatPoolSkip::Retryable)
));
let identity = pool_meta_with_decommission(PoolDecommissionInfo::default());
assert!(matches!(
classify_heal_format_pool(0, "pool-new", &identity, None),
Some(HealFormatPoolSkip::Retryable)
));
let identity_without_decommission = PoolMeta {
pools: vec![PoolStatus {
id: 0,
cmd_line: "pool-0".to_string(),
last_update: OffsetDateTime::UNIX_EPOCH,
decommission: None,
}],
..Default::default()
};
assert!(matches!(
classify_heal_format_pool(0, "pool-new", &identity_without_decommission, None),
Some(HealFormatPoolSkip::Retryable)
));
assert!(matches!(
classify_heal_format_pool(0, "", &identity_meta, None),
Some(HealFormatPoolSkip::Retryable)
));
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &PoolMeta::default(), None),
Some(HealFormatPoolSkip::Retryable)
));
let stopped = RebalanceMeta {
stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
pool_stats: vec![RebalanceStats {
participating: true,
info: RebalanceInfo {
status: RebalStatus::Stopped,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
assert!(classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopped)).is_none());
let stopping_after_stop = RebalanceMeta {
stopped_at: Some(OffsetDateTime::UNIX_EPOCH),
pool_stats: vec![RebalanceStats {
participating: true,
info: RebalanceInfo {
status: RebalStatus::Started,
stopping: true,
..Default::default()
},
..Default::default()
}],
..Default::default()
};
assert!(matches!(
classify_heal_format_pool(0, "pool-0", &identity_meta, Some(&stopping_after_stop)),
Some(HealFormatPoolSkip::Retryable)
));
}
#[test]
fn skipped_heal_format_pool_is_never_reported_as_success() {
assert!(matches!(
heal_format_pool_skip_error(HealFormatPoolSkip::Retryable),
StorageError::SlowDown
));
assert!(matches!(
heal_format_pool_skip_error(HealFormatPoolSkip::Completed),
StorageError::NoHealRequired
));
}
async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) { async fn multi_pool_heal_store() -> (tempfile::TempDir, Arc<ECStore>, CancellationToken) {
let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created"); let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
let mut pool_endpoints = Vec::new(); let mut pool_endpoints = Vec::new();
@@ -889,6 +1189,18 @@ mod tests {
bucket_fence_registry: std::sync::Arc::default(), bucket_fence_registry: std::sync::Arc::default(),
}; };
let err = store
.handle_heal_format(false)
.await
.expect_err("missing pool metadata must fail closed before format writes");
assert!(matches!(err, StorageError::SlowDown));
let pool_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
pool_meta
.save(store.pools.clone())
.await
.expect("pool metadata should be persisted before format heal");
let (result, err) = store let (result, err) = store
.handle_heal_format(false) .handle_heal_format(false)
.await .await
@@ -902,5 +1214,22 @@ mod tests {
.await .await
.expect("the later pool should be healed despite the first pool error"); .expect("the later pool should be healed despite the first pool error");
assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]); assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]);
let mut completed_meta = PoolMeta::new(&store.pools, &PoolMeta::default());
for status in &mut completed_meta.pools {
status.decommission = Some(PoolDecommissionInfo {
complete: true,
..Default::default()
});
}
completed_meta
.save(store.pools.clone())
.await
.expect("completed pool metadata should be persisted");
let (_, err) = store
.handle_heal_format(false)
.await
.expect("completed pools should be reported as a no-op");
assert!(matches!(err, Some(StorageError::NoHealRequired)));
} }
} }
+1 -1
View File
@@ -3194,7 +3194,7 @@ impl ECStore {
// Default return value // Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()]; let mut del_objects = vec![DeletedObject::default(); objects.len()];
let mut accounting = vec![None; objects.len()]; let accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len()); let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..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)) .filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx)); latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
let Some(winner) = latest_candidates.first() else { let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum); return Err(Error::ErasureReadQuorum);
@@ -231,6 +231,10 @@ impl HealTask {
"Heal erasure set format repair skipped because no format heal was required" "Heal erasure set format repair skipped because no format heal was required"
); );
} else { } else {
let error = e;
if error.is_recoverable_heal() {
return Err(error);
}
error!( error!(
target: "rustfs::heal::task", target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_RESULT, event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -239,7 +243,7 @@ impl HealTask {
task_id = %self.id, task_id = %self.id,
set_disk_id, set_disk_id,
result = "format_failed", result = "format_failed",
error = %e, error = %error,
"Heal erasure set failed" "Heal erasure set failed"
); );
{ {
@@ -247,7 +251,7 @@ impl HealTask {
progress.update_progress(4, 4, 0, 0); progress.update_progress(4, 4, 0, 0);
} }
return Err(Error::TaskExecutionFailed { return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal disk format for {set_disk_id}: {e}"), message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
}); });
} }
} else { } else {
@@ -284,6 +288,9 @@ impl HealTask {
Err(Error::TaskCancelled) => return Err(Error::TaskCancelled), Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
Err(Error::TaskTimeout) => return Err(Error::TaskTimeout), Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
Err(e) => { Err(e) => {
if e.is_recoverable_heal() {
return Err(e);
}
error!( error!(
target: "rustfs::heal::task", target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_RESULT, event = EVENT_HEAL_ERASURE_SET_RESULT,
+28
View File
@@ -547,6 +547,7 @@ struct MockStorage {
heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>, heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>, heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
format_no_heal_required: Mutex<bool>, format_no_heal_required: Mutex<bool>,
format_error: Mutex<Option<Error>>,
global_format_calls: Mutex<u32>, global_format_calls: Mutex<u32>,
replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>, replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
replacement_targets_ready: Mutex<bool>, replacement_targets_ready: Mutex<bool>,
@@ -867,6 +868,9 @@ impl HealStorageAPI for MockStorage {
async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> { async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
*self.global_format_calls.lock().unwrap() += 1; *self.global_format_calls.lock().unwrap() += 1;
if let Some(error) = self.format_error.lock().unwrap().take() {
return Err(error);
}
let no_heal_required = *self.format_no_heal_required.lock().unwrap(); let no_heal_required = *self.format_no_heal_required.lock().unwrap();
if no_heal_required { if no_heal_required {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired)))) Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2052,6 +2056,30 @@ async fn test_erasure_set_heal_continues_after_format_no_heal_required() {
); );
} }
#[tokio::test]
async fn erasure_set_format_slowdown_is_propagated() {
let storage = Arc::new(MockStorage {
format_error: Mutex::new(Some(Error::Storage(EcstoreError::SlowDown))),
..Default::default()
});
let request = HealRequest::new(
HealType::ErasureSet {
buckets: Vec::new(),
set_disk_id: "pool_0_set_0".to_string(),
},
HealOptions::default(),
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage);
let error = task
.execute()
.await
.expect_err("format SlowDown must remain recoverable for the task manager");
assert!(matches!(error, Error::Storage(EcstoreError::SlowDown)));
}
#[tokio::test] #[tokio::test]
async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() { async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
let temp = TempDir::new().expect("temporary directory should be created"); let temp = TempDir::new().expect("temporary directory should be created");
+86 -25
View File
@@ -16,14 +16,14 @@
//! //!
//! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
//! kills the active node while this test continuously decrypts through a //! kills the active node while this test continuously decrypts through a
//! surviving standby. KV2 and Transit requests must remain successful, use a //! surviving standby. KV2 and Transit must recover after the bounded circuit
//! bounded number of attempts, and leave the circuit and in-flight gauges at //! interval, use a bounded number of attempts, and leave the circuit and
//! zero after a new leader is elected. //! in-flight gauges at zero after a new leader is elected.
use std::collections::HashMap; use std::collections::HashMap;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
use metrics_util::MetricKind; use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight"; const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open"; const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
const MAX_ATTEMPTS: u32 = 10; 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 = ( type MetricEntry = (
metrics_util::CompositeKey, metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
backend, backend,
backend_config, backend_config,
allow_insecure_dev_defaults: true, allow_insecure_dev_defaults: true,
timeout: Duration::from_secs(2), timeout: ATTEMPT_TIMEOUT,
retry_attempts: MAX_ATTEMPTS, retry_attempts: MAX_ATTEMPTS,
enable_cache: false, enable_cache: false,
..KmsConfig::default() ..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
.sum() .sum()
} }
async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) { async fn wait_for_count(
tokio::time::timeout(Duration::from_secs(20), async { counter: &AtomicU64,
failure: &Mutex<Option<String>>,
minimum: u64,
description: &str,
timeout: Duration,
) {
tokio::time::timeout(timeout, async {
while counter.load(Ordering::SeqCst) < minimum { 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; tokio::time::sleep(Duration::from_millis(25)).await;
} }
}) })
.await .await
.unwrap_or_else(|_| panic!("timed out waiting for {description}")); .unwrap_or_else(|_| {
panic!(
"timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
counter.load(Ordering::SeqCst)
)
});
} }
async fn wait_for_file(path: &Path, description: &str) { async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
request: DecryptRequest, request: DecryptRequest,
expected: Vec<u8>, expected: Vec<u8>,
completed: Arc<AtomicU64>, completed: Arc<AtomicU64>,
failed: Arc<AtomicBool>, allow_failover_errors: Arc<AtomicBool>,
failure: Arc<Mutex<Option<String>>>,
stop: CancellationToken, stop: CancellationToken,
) { ) {
while !stop.is_cancelled() { while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
Ok(response) if response.plaintext == expected => { Ok(response) if response.plaintext == expected => {
completed.fetch_add(1, Ordering::SeqCst); completed.fetch_add(1, Ordering::SeqCst);
} }
Ok(_) | Err(_) => { Ok(_) => {
failed.store(true, Ordering::SeqCst); *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());
return; return;
} }
} }
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
); );
let stop = CancellationToken::new(); let stop = CancellationToken::new();
let failed = Arc::new(AtomicBool::new(false)); 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 kv2_completed = Arc::new(AtomicU64::new(0)); let kv2_completed = Arc::new(AtomicU64::new(0));
let transit_completed = Arc::new(AtomicU64::new(0)); let transit_completed = Arc::new(AtomicU64::new(0));
let kv2_worker = tokio::spawn(decrypt_loop( let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
kv2_request, kv2_request,
kv2_data_key.plaintext_key, kv2_data_key.plaintext_key,
Arc::clone(&kv2_completed), Arc::clone(&kv2_completed),
Arc::clone(&failed), Arc::clone(&allow_failover_errors),
Arc::clone(&kv2_failure),
stop.clone(), stop.clone(),
)); ));
let transit_worker = tokio::spawn(decrypt_loop( let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
transit_request, transit_request,
transit_data_key.plaintext_key, transit_data_key.plaintext_key,
Arc::clone(&transit_completed), Arc::clone(&transit_completed),
Arc::clone(&failed), Arc::clone(&allow_failover_errors),
Arc::clone(&transit_failure),
stop.clone(), stop.clone(),
)); ));
wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await; wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await; wait_for_count(
&transit_completed,
&transit_failure,
2,
"two healthy Transit decrypts",
HEALTHY_PROGRESS_TIMEOUT,
)
.await;
allow_failover_errors.store(true, Ordering::SeqCst);
std::fs::write(&marker, b"ready").expect("publish failover readiness marker"); std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
wait_for_file(&elected, "the replacement Vault leader").await; wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2; let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2; let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await; wait_for_count(
wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await; &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;
stop.cancel(); stop.cancel();
kv2_worker.await.expect("KV2 decrypt worker must join"); kv2_worker.await.expect("KV2 decrypt worker must join");
transit_worker.await.expect("Transit decrypt worker must join"); transit_worker.await.expect("Transit decrypt worker must join");
assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext"); 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"
);
} }
#[test] #[test]
#[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"] #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() { fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
let recorder = DebuggingRecorder::new(); let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter(); let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || { metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
}); });
let snapshot = snapshotter.snapshot().into_vec(); 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!( assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
0, 0,
+33 -4
View File
@@ -54,6 +54,7 @@ const ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS: &str =
"Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions"; "Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions";
const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration"; const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration";
const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters"; const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
const ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY: &str = "Rule ID must not be empty";
const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled"; const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering"; const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering";
const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str = const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
@@ -402,10 +403,13 @@ impl Lifecycle for BucketLifecycleConfiguration {
NoncurrentVersionTransitionOps::validate(transition)?; NoncurrentVersionTransitionOps::validate(transition)?;
} }
} }
if let Some(id) = &r.id if let Some(id) = &r.id {
&& id.len() > 255 if id.is_empty() {
{ return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY));
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG)); }
if id.len() > 255 {
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
}
} }
r.validate()?; r.validate()?;
if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref() if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref()
@@ -3730,6 +3734,31 @@ mod tests {
.expect("empty prefix with filter should be valid"); .expect("empty prefix with filter should be valid");
} }
#[tokio::test]
async fn validate_rejects_empty_rule_id() {
let lc = BucketLifecycleConfiguration {
expiry_updated_at: None,
rules: vec![LifecycleRule {
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
expiration: Some(LifecycleExpiration {
days: Some(30),
..Default::default()
}),
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
filter: None,
id: Some(String::new()),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
};
let error = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
assert_eq!(error.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY);
}
// --- TASK-004 tests: ExpiredObjectAllVersions --- // --- TASK-004 tests: ExpiredObjectAllVersions ---
#[tokio::test] #[tokio::test]
+50 -5
View File
@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
pub use rustfs_data_usage::{ pub use rustfs_data_usage::{
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME, AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache, PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry, SizeReconciliationScope, SizeSummary,
TierStats, hash_path, prefix_usage_in_cache,
}; };
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout}; use tokio::time::{Duration, Instant, sleep, timeout};
@@ -192,6 +193,10 @@ const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
pub trait ScannerSizeSummaryExt { pub trait ScannerSizeSummaryExt {
/// Fold one object's contribution into the summary, including its tier. /// Fold one object's contribution into the summary, including its tier.
fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64); fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
/// Fold counters and physical tier usage for an object whose metadata is
/// valid but whose logical size is currently unavailable. Logical totals
/// stay unchanged.
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo);
} }
impl ScannerSizeSummaryExt for SizeSummary { impl ScannerSizeSummaryExt for SizeSummary {
@@ -225,6 +230,34 @@ impl ScannerSizeSummaryExt for SizeSummary {
}); });
} }
} }
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo) {
if oi.delete_marker {
self.delete_markers = self.delete_markers.saturating_add(1);
return;
}
if oi.version_id.is_some_and(|v| !v.is_nil()) {
self.versions = self.versions.saturating_add(1);
}
if oi.transitioned_object.free_version {
return;
}
let tier = if oi.transitioned_object.status == TRANSITION_COMPLETE {
oi.transitioned_object.tier.clone()
} else {
oi.storage_class.clone().unwrap_or_else(|| storageclass::STANDARD.to_string())
};
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),
});
}
}
} }
// ===== Cache-related data structures ===== // ===== Cache-related data structures =====
@@ -344,6 +377,10 @@ pub struct DataUsageCacheInfo {
pub scan_plan_digest: Option<DataUsageScanPlanDigest>, pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
#[serde(default)] #[serde(default)]
pub cache_key_format: u16, pub cache_key_format: u16,
/// Bounded durable debts for versions whose logical size was not trusted.
/// The map key is an identity key, never a user-controlled metric label.
#[serde(default)]
pub size_reconciliation: HashMap<String, SizeReconciliationEntry>,
} }
impl Serialize for DataUsageCacheInfo { impl Serialize for DataUsageCacheInfo {
@@ -353,7 +390,8 @@ impl Serialize for DataUsageCacheInfo {
{ {
// Keep this metadata map-encoded so older readers can ignore fields // Keep this metadata map-encoded so older readers can ignore fields
// appended by newer scanner versions during rolling upgrades. // appended by newer scanner versions during rolling upgrades.
let mut state = serializer.serialize_map(Some(16))?; let field_count = 16 + usize::from(!self.size_reconciliation.is_empty());
let mut state = serializer.serialize_map(Some(field_count))?;
state.serialize_entry("name", &self.name)?; state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?; state.serialize_entry("next_cycle", &self.next_cycle)?;
state.serialize_entry("leader_epoch", &self.leader_epoch)?; state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -370,6 +408,9 @@ impl Serialize for DataUsageCacheInfo {
state.serialize_entry("snapshot_complete", &self.snapshot_complete)?; state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?; state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
state.serialize_entry("cache_key_format", &self.cache_key_format)?; state.serialize_entry("cache_key_format", &self.cache_key_format)?;
if !self.size_reconciliation.is_empty() {
state.serialize_entry("size_reconciliation", &self.size_reconciliation)?;
}
state.end() state.end()
} }
} }
@@ -428,14 +469,18 @@ impl DataUsageCache {
self.checked_flatten(name).is_some() self.checked_flatten(name).is_some()
}); });
if !reusable { if !reusable {
let pending_heals = if self.info.name == name { let (pending_heals, size_reconciliation) = if self.info.name == name {
std::mem::take(&mut self.info.pending_heals) (
std::mem::take(&mut self.info.pending_heals),
std::mem::take(&mut self.info.size_reconciliation),
)
} else { } else {
Vec::new() (Vec::new(), HashMap::new())
}; };
*self = Self::default(); *self = Self::default();
self.info.name = name.to_string(); self.info.name = name.to_string();
self.info.pending_heals = pending_heals; self.info.pending_heals = pending_heals;
self.info.size_reconciliation = size_reconciliation;
} }
self.info.next_cycle = next_cycle; self.info.next_cycle = next_cycle;
@@ -673,6 +673,34 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
); );
} }
#[test]
fn size_summary_unknown_accounting_keeps_physical_tier_and_version_only() {
let mut summary = SizeSummary::default();
summary
.tier_stats
.insert(storageclass::STANDARD.to_string(), TierStats::default());
let object = ObjectInfo {
size: 12,
storage_class: Some(storageclass::STANDARD.to_string()),
version_id: Some(uuid::Uuid::new_v4()),
is_latest: true,
..Default::default()
};
summary.actions_accounting_unknown(&object);
assert_eq!(summary.total_size, 0, "unknown logical size must not become zero or physical bytes");
assert_eq!(summary.versions, 1);
assert_eq!(
summary.tier_stats.get(storageclass::STANDARD),
Some(&TierStats {
total_size: 12,
num_versions: 1,
num_objects: 1,
})
);
}
#[test] #[test]
fn test_data_usage_entry_merge_sums_failed_objects() { fn test_data_usage_entry_merge_sums_failed_objects() {
let mut left = DataUsageEntry { let mut left = DataUsageEntry {
@@ -1079,6 +1107,16 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
scan_plan_digest: Some(TEST_PLAN_DIGEST), scan_plan_digest: Some(TEST_PLAN_DIGEST),
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT, cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
pending_heals: vec![pending_heal.clone()], pending_heals: vec![pending_heal.clone()],
size_reconciliation: HashMap::from([(
"size-key".to_string(),
SizeReconciliationEntry {
key: "size-key".to_string(),
bucket: "bucket".to_string(),
object: "prefix/object".to_string(),
reason: "invalid_declared_size".to_string(),
..Default::default()
},
)]),
..Default::default() ..Default::default()
}, },
..Default::default() ..Default::default()
@@ -1088,6 +1126,7 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
assert_eq!(outcome, DataUsageCachePrepareOutcome::Reused); assert_eq!(outcome, DataUsageCachePrepareOutcome::Reused);
assert_eq!(cache.info.pending_heals, vec![pending_heal]); assert_eq!(cache.info.pending_heals, vec![pending_heal]);
assert!(cache.info.size_reconciliation.contains_key("size-key"));
assert!(cache.cache.is_empty()); assert!(cache.cache.is_empty());
assert!(!cache.info.snapshot_complete); assert!(!cache.info.snapshot_complete);
} }
+129 -5
View File
@@ -20,8 +20,9 @@ use std::time::{Duration, Instant, SystemTime};
use crate::ReplTargetSizeSummary; use crate::ReplTargetSizeSummary;
use crate::data_usage_define::{ use crate::data_usage_define::{
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint, DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
SizeReconciliationEntry, SizeSummary, hash_path,
}; };
use crate::error::ScannerError; use crate::error::ScannerError;
use crate::runtime_config::{ use crate::runtime_config::{
@@ -97,6 +98,9 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total"; const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total"; const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128; const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
const MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET: usize = 10_000;
const MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET: usize = 8 * 1024 * 1024;
const MAX_SIZE_RECONCILIATION_AGE_SECS: u64 = 7 * 24 * 60 * 60;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) -- // --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
// //
@@ -364,7 +368,7 @@ impl PendingScannerAccounting<'_> {
fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) { fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) {
let size = if queued { self.expired_size } else { self.retained_size }; let size = if queued { self.expired_size } else { self.retained_size };
size_summary.actions_accounting(self.object, size, self.retained_size); size_summary.actions_accounting(self.object, size, self.retained_size);
*cumulative_size += size; *cumulative_size = cumulative_size.saturating_add(size);
} }
} }
@@ -671,10 +675,65 @@ pub struct FolderScanner {
skip_heal: Arc<std::sync::atomic::AtomicBool>, skip_heal: Arc<std::sync::atomic::AtomicBool>,
local_disk: Arc<Disk>, local_disk: Arc<Disk>,
pending_heals_changed: bool, pending_heals_changed: bool,
pending_size_reconciliation_keys: HashSet<String>,
pending_size_reconciliation_scopes: HashSet<String>,
pending_size_reconciliation_truncated: bool,
#[cfg(test)] #[cfg(test)]
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>, list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
} }
fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
entry.key.len()
+ entry.bucket.len()
+ entry.object.len()
+ entry.version_id.as_deref().map_or(0, str::len)
+ entry.generation.as_deref().map_or(0, str::len)
+ entry.reason.len()
+ std::mem::size_of::<u64>()
+ std::mem::size_of::<u32>()
}
fn size_reconciliation_scope_key(bucket: &str, object: &str) -> String {
format!("{}:{}|{}:{}", bucket.len(), bucket, object.len(), object)
}
fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
info.size_reconciliation.retain(|key, entry| {
if entry.first_seen == 0 || entry.first_seen > now {
entry.first_seen = now;
}
key == &entry.key
&& entry.key.len() <= 4096
&& entry.bucket.len() <= 512
&& entry.object.len() <= 512
&& entry.version_id.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.generation.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.reason.len() <= 64
&& now.saturating_sub(entry.first_seen) <= MAX_SIZE_RECONCILIATION_AGE_SECS
});
while info.size_reconciliation.len() > MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET
|| info
.size_reconciliation
.values()
.map(size_reconciliation_entry_bytes)
.sum::<usize>()
> MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET
{
let oldest = info
.size_reconciliation
.iter()
.min_by(|(left_key, left), (right_key, right)| {
left.first_seen.cmp(&right.first_seen).then_with(|| left_key.cmp(right_key))
})
.map(|(key, _)| key.clone());
let Some(oldest) = oldest else {
break;
};
info.size_reconciliation.remove(&oldest);
}
}
impl FolderScanner { impl FolderScanner {
fn now_secs() -> u64 { fn now_secs() -> u64 {
SystemTime::now() SystemTime::now()
@@ -748,6 +807,60 @@ impl FolderScanner {
} }
} }
/// Apply the per-object size-resolution ledger updates in one place. The
/// scanner cache is the durable boundary; both working copies are updated
/// so an incremental publication cannot lose a debt or its resolution.
fn apply_size_reconciliation(&mut self, summary: &SizeSummary) {
let now = Self::now_secs();
self.pending_size_reconciliation_keys
.extend(summary.size_reconciliation.iter().map(|entry| entry.key.clone()));
self.pending_size_reconciliation_scopes.extend(
summary
.reconciliation_scopes
.iter()
.map(|scope| size_reconciliation_scope_key(&scope.bucket, &scope.object)),
);
self.pending_size_reconciliation_truncated |= summary.size_reconciliation_truncated;
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
for incoming in &summary.size_reconciliation {
if let Some(existing) = info.size_reconciliation.get_mut(&incoming.key) {
existing.reason = incoming.reason.clone();
existing.physical_size = incoming.physical_size;
existing.generation = incoming.generation.clone();
existing.version_id = incoming.version_id.clone();
existing.attempts = existing.attempts.saturating_add(1);
continue;
}
if size_reconciliation_entry_bytes(incoming) > MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET {
continue;
}
let mut entry = incoming.clone();
entry.first_seen = now;
entry.attempts = 1;
info.size_reconciliation.insert(entry.key.clone(), entry);
}
}
}
fn finish_size_reconciliation_batch(&mut self) {
let now = Self::now_secs();
let current_keys = std::mem::take(&mut self.pending_size_reconciliation_keys);
let scopes = std::mem::take(&mut self.pending_size_reconciliation_scopes);
let truncated = std::mem::replace(&mut self.pending_size_reconciliation_truncated, false);
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
if !truncated {
info.size_reconciliation.retain(|key, entry| {
!scopes.contains(&size_reconciliation_scope_key(&entry.bucket, &entry.object)) || current_keys.contains(key)
});
}
prune_size_reconciliation(info, now);
}
}
fn record_scan_resume_hint(&mut self, folder: &str) { fn record_scan_resume_hint(&mut self, folder: &str) {
self.new_cache.info.scan_resume_after = Some(folder.to_string()); self.new_cache.info.scan_resume_after = Some(folder.to_string());
self.update_cache.info.scan_resume_after = Some(folder.to_string()); self.update_cache.info.scan_resume_after = Some(folder.to_string());
@@ -1426,6 +1539,7 @@ impl FolderScanner {
abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()])); abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
apply_scanner_size_summary(into, &sz); apply_scanner_size_summary(into, &sz);
self.apply_size_reconciliation(&sz);
into.objects += 1; into.objects += 1;
object_count += 1; object_count += 1;
self.budget.record_object_scanned(); self.budget.record_object_scanned();
@@ -2105,6 +2219,7 @@ impl FolderScanner {
} }
} }
self.finish_size_reconciliation_batch();
done_folder(); done_folder();
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX); let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed"); emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
@@ -2190,10 +2305,17 @@ pub async fn scan_data_folder(
skip_heal, skip_heal,
local_disk, local_disk,
pending_heals_changed: false, pending_heals_changed: false,
pending_size_reconciliation_keys: HashSet::new(),
pending_size_reconciliation_scopes: HashSet::new(),
pending_size_reconciliation_truncated: false,
#[cfg(test)] #[cfg(test)]
list_path_raw_options_observer: None, list_path_raw_options_observer: None,
}; };
let now = FolderScanner::now_secs();
prune_size_reconciliation(&mut scanner.new_cache.info, now);
prune_size_reconciliation(&mut scanner.update_cache.info, now);
// Check if context is cancelled // Check if context is cancelled
if ctx.is_cancelled() { if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string())); return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -2217,7 +2339,9 @@ pub async fn scan_data_folder(
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
new_cache.info.last_update = Some(SystemTime::now()); new_cache.info.last_update = Some(SystemTime::now());
new_cache.info.next_cycle = cache.info.next_cycle; new_cache.info.next_cycle = cache.info.next_cycle;
let unresolved_objects = root.failed_objects > 0 || !new_cache.info.failed_objects.is_empty(); let unresolved_objects = root.failed_objects > 0
|| !new_cache.info.failed_objects.is_empty()
|| !new_cache.info.size_reconciliation.is_empty();
new_cache.info.snapshot_complete = !unresolved_objects; new_cache.info.snapshot_complete = !unresolved_objects;
let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some(); let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
new_cache.info.scan_resume_after = None; new_cache.info.scan_resume_after = None;
@@ -2245,7 +2369,7 @@ pub async fn scan_data_folder(
if root_has_progress { if root_has_progress {
new_cache.replace_hashed(&root_hash, &None, &root); new_cache.replace_hashed(&root_hash, &None, &root);
} }
if partial_cache_is_useful(&root, pending_heals_changed) { if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() {
if new_cache.root().is_some() { if new_cache.root().is_some() {
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
} }
+809 -65
View File
@@ -13,6 +13,7 @@
// limitations under the License. // limitations under the License.
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers. /// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
use super::*; use super::*;
use sha2::{Digest as _, Sha256};
/// Cached folder information for scanning /// Cached folder information for scanning
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
@@ -32,6 +33,263 @@ pub(super) enum GetSizeFailureAction {
HealMetadata { object: String }, HealMetadata { object: String },
} }
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SizeResolutionReason {
CompressedSizeUnknown,
InvalidPhysicalSize,
UnsupportedCompression,
InvalidObjectSize,
InvalidPartSize,
InvalidDeclaredSize,
SizeOverflowOrMismatch,
}
impl SizeResolutionReason {
fn as_str(self) -> &'static str {
match self {
Self::CompressedSizeUnknown => "compressed_size_unknown",
Self::InvalidPhysicalSize => "invalid_physical_size",
Self::UnsupportedCompression => "unsupported_compression",
Self::InvalidObjectSize => "invalid_object_size",
Self::InvalidPartSize => "invalid_part_size",
Self::InvalidDeclaredSize => "invalid_declared_size",
Self::SizeOverflowOrMismatch => "size_overflow_or_mismatch",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum SizeResolution {
Known { logical: i64, physical: i64 },
Unknown { physical: i64, reason: SizeResolutionReason },
Corrupt { physical: i64, reason: SizeResolutionReason },
}
impl SizeResolution {
fn known_size(&self) -> Option<i64> {
match self {
Self::Known { logical, .. } => Some(*logical),
Self::Unknown { .. } | Self::Corrupt { .. } => None,
}
}
}
fn size_reconciliation_key(oi: &ObjectInfo, reason: SizeResolutionReason) -> String {
let version = oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string())
.unwrap_or_default();
let generation = oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string())
.unwrap_or_default();
// Length-prefix each component so an object key containing the separator
// cannot alias another identity. S3 keys are bounded in normal operation;
// oversized persisted values use a digest so a corrupt metadata record
// cannot grow the ledger without bound.
fn component(value: &str) -> String {
const MAX_COMPONENT_LEN: usize = 512;
if value.len() <= MAX_COMPONENT_LEN {
return format!("{}:{}", value.len(), value);
}
let digest = Sha256::digest(value.as_bytes());
let digest = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
format!("hash:{}:{}", value.len(), digest)
}
format!(
"{}|{}|{}|{}|{}",
component(&oi.bucket),
component(&oi.name),
component(&version),
component(&generation),
component(reason.as_str())
)
}
pub(super) fn bounded_reconciliation_field(value: &str) -> String {
const MAX_FIELD_LEN: usize = 512;
if value.len() <= MAX_FIELD_LEN {
return value.to_string();
}
let digest = hex_simd::encode_to_string(Sha256::digest(value.as_bytes()), hex_simd::AsciiCase::Lower);
let prefix_len = MAX_FIELD_LEN - 65;
let prefix = value
.char_indices()
.take_while(|(offset, ch)| offset.saturating_add(ch.len_utf8()) <= prefix_len)
.map(|(_, ch)| ch)
.collect::<String>();
format!("{}~{}", prefix, digest)
}
fn record_size_resolution(summary: &mut SizeSummary, oi: &ObjectInfo, resolution: &SizeResolution) {
match resolution {
SizeResolution::Known { .. } => {}
SizeResolution::Unknown { physical, reason } | SizeResolution::Corrupt { physical, reason } => {
summary.record_size_reconciliation(SizeReconciliationEntry {
key: size_reconciliation_key(oi, *reason),
bucket: bounded_reconciliation_field(&oi.bucket),
object: bounded_reconciliation_field(&oi.name),
version_id: oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string()),
generation: oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string()),
reason: reason.as_str().to_string(),
physical_size: u64::try_from(*physical).ok(),
first_seen: 0,
attempts: 0,
});
}
}
}
/// Resolve the size metadata once at the scanner trust boundary. A compressed
/// -1 sentinel is valid legacy metadata, but it cannot participate in normal
/// logical-size accounting or size-filtered lifecycle rules.
pub(super) fn resolve_size(oi: &ObjectInfo) -> SizeResolution {
let physical = oi.size;
if physical < 0 {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPhysicalSize,
};
}
let compressed = match oi.compression_read_plan() {
Ok((_, _, compressed)) => compressed,
Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::UnsupportedCompression,
};
}
};
if oi.actual_size < -1 || (oi.actual_size == -1 && !compressed) {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidObjectSize,
};
}
// Match ObjectInfo::get_actual_size: a positive in-memory value is the
// authoritative decoded size. Stale declared/part metadata must not turn
// an otherwise valid object into a false corruption report.
if oi.actual_size > 0 {
return SizeResolution::Known {
logical: oi.actual_size,
physical,
};
}
if oi
.parts
.iter()
.any(|part| part.actual_size < -1 || (part.actual_size < 0 && !compressed))
{
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPartSize,
};
}
let declared = rustfs_utils::http::get_str(&oi.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
let declared = match declared {
Some(value) if value.is_empty() => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
Some(value) => match value.parse::<i64>() {
Ok(value) if value >= 0 => Some(value),
_ => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
},
None => None,
};
let logical = match oi.get_actual_size() {
Ok(size) if size == -1 && compressed && declared.is_none() => {
return SizeResolution::Unknown {
physical,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
}
Ok(size) if size >= 0 => size,
Ok(_) | Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
};
if compressed && logical == 0 && physical != 0 && oi.parts.is_empty() && declared.is_none() {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
SizeResolution::Known { logical, physical }
}
fn resolve_sizes(object_infos: &[ObjectInfo]) -> Vec<SizeResolution> {
object_infos.iter().map(resolve_size).collect()
}
fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule_id: &str) -> bool {
let filter_has_size = |filter: &s3s::dto::LifecycleRuleFilter| {
filter.object_size_greater_than.is_some()
|| filter.object_size_less_than.is_some()
|| filter
.and
.as_ref()
.is_some_and(|and| and.object_size_greater_than.is_some() || and.object_size_less_than.is_some())
};
lifecycle
.rules
.iter()
.find(|rule| {
if rule_id.is_empty() {
rule.id.as_deref().is_none_or(str::is_empty)
} else {
rule.id.as_deref() == Some(rule_id)
}
})
.and_then(|rule| rule.filter.as_ref())
.is_some_and(filter_has_size)
}
fn lifecycle_event_allowed(resolution: &SizeResolution, event: &Event, lifecycle: &BucketLifecycleConfiguration) -> bool {
match resolution {
// Missing or invalid logical size only defers actions whose selected
// rule actually depends on that size. Time/version-only actions retain
// their existing semantics, including intrinsic events without a rule ID.
SizeResolution::Unknown { .. } | SizeResolution::Corrupt { .. } => {
!lifecycle_rule_has_size_filter(lifecycle, &event.rule_id)
}
SizeResolution::Known { .. } => true,
}
}
/// A successful newer-noncurrent batch consumes both known and unresolved
/// versions from the retained-version alert count. The two accounting paths
/// are separate because only known sizes can contribute byte totals.
fn remaining_versions_after_queued_noncurrent(remaining_versions: usize, known_count: usize, unknown_count: usize) -> usize {
remaining_versions.saturating_sub(known_count.saturating_add(unknown_count))
}
/// How the corrupt-metadata branch records the repair after attempting an /// How the corrupt-metadata branch records the repair after attempting an
/// MRF intent (backlog#1894 axis A). /// MRF intent (backlog#1894 axis A).
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
@@ -319,34 +577,48 @@ impl ScannerItem {
"Scanner lifecycle evaluation started" "Scanner lifecycle evaluation started"
); );
let resolved_sizes = resolve_sizes(&object_infos);
if let Some(first) = object_infos.first() {
size_summary.record_reconciliation_scope(
&bounded_reconciliation_field(&first.bucket),
&bounded_reconciliation_field(&first.name),
);
}
for (oi, resolution) in object_infos.iter().zip(resolved_sizes.iter()) {
record_size_resolution(size_summary, oi, resolution);
}
let has_corrupt_size = resolved_sizes
.iter()
.any(|resolution| matches!(resolution, SizeResolution::Corrupt { .. }));
// `versioning_config` is resolved once per object by the caller // `versioning_config` is resolved once per object by the caller
// (`get_size`) and handed in; only `prefix_enabled` is consulted here. // (`get_size`) and handed in; only `prefix_enabled` is consulted here.
let Some(lifecycle) = self.lifecycle.as_ref() else { let Some(lifecycle) = self.lifecycle.clone() else {
let mut cumulative_size = 0; let mut cumulative_size: i64 = 0;
for oi in object_infos.iter() { for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
let actual_size = match oi.get_actual_size() { let accounting_size = match resolved_size {
Ok(size) => size, SizeResolution::Known { logical, .. } => *logical,
Err(_) => { // A valid compressed legacy sentinel has no logical size,
warn!( // but heal and replication still need to run. The
target: "rustfs::scanner::folder", // physical size is only an input to those operations; it
event = EVENT_SCANNER_LIFECYCLE_ACTION, // is not folded into the logical total below.
component = LOG_COMPONENT_SCANNER, SizeResolution::Unknown { physical, .. } => {
subsystem = LOG_SUBSYSTEM_LIFECYCLE, self.heal_actions(oi, *physical, size_summary).await;
bucket = %self.bucket, size_summary.actions_accounting_unknown(oi);
object = %oi.name, continue;
state = "size_lookup_failed", }
"Scanner lifecycle action used fallback size" SizeResolution::Corrupt { .. } => {
); size_summary.actions_accounting_unknown(oi);
continue; continue;
} }
}; };
let size = self.heal_actions(oi, actual_size, size_summary).await; let size = self.heal_actions(oi, accounting_size, size_summary).await;
size_summary.actions_accounting(oi, size, actual_size); size_summary.actions_accounting(oi, size, accounting_size);
cumulative_size += size; cumulative_size = cumulative_size.saturating_add(size);
} }
self.alert_excessive_versions(object_infos.len(), cumulative_size); self.alert_excessive_versions(object_infos.len(), cumulative_size);
@@ -400,25 +672,108 @@ impl ScannerItem {
let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new(); let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
let mut noncurrent_events: Vec<Event> = Vec::new(); let mut noncurrent_events: Vec<Event> = Vec::new();
let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new(); let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new();
let mut noncurrent_unknown: Vec<&ObjectInfo> = Vec::new();
let mut cumulative_size = 0; let mut cumulative_size = 0;
let mut remaining_versions = object_infos.len(); let mut remaining_versions = object_infos.len();
'eventLoop: { 'eventLoop: {
for (i, event) in events.iter().enumerate() { for (i, event) in events.iter().enumerate() {
let oi = &object_infos[i]; let oi = &object_infos[i];
let actual_size = match oi.get_actual_size() { let known_size = resolved_sizes[i].known_size();
Ok(size) => size, if has_corrupt_size
Err(_) => { && matches!(
warn!( event.action,
target: "rustfs::scanner::folder", IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction
event = EVENT_SCANNER_LIFECYCLE_ACTION, )
component = LOG_COMPONENT_SCANNER, {
subsystem = LOG_SUBSYSTEM_LIFECYCLE, // An all-version delete would also remove a corrupt
bucket = %self.bucket, // sibling that could not be reconciled safely.
object = %oi.name, continue;
state = "size_lookup_failed", }
"Scanner lifecycle action used fallback size" if !lifecycle_event_allowed(&resolved_sizes[i], event, &lifecycle) {
); // An unknown logical size must not make an otherwise
0 // non-destructive scan disappear from heal/physical-tier
// accounting. Size-filtered or deferred events remain
// pending, so retain the version-only physical counters.
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
size_summary.actions_accounting_unknown(oi);
}
continue;
}
let actual_size = match known_size {
Some(size) => size,
None => {
match event.action {
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction => {
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
if event.action == IlmAction::DeleteAllVersionsAction
|| event.action == IlmAction::DelMarkerDeleteAllVersionsAction
{
remaining_versions = 0;
}
} else if matches!(
event.action,
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
) {
size_summary.actions_accounting_unknown(oi);
} else {
size_summary.actions_accounting_unknown(oi);
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
match &resolved_sizes[j] {
SizeResolution::Known { logical, .. } => PendingScannerAccounting {
object: retained,
retained_size: *logical,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false),
SizeResolution::Unknown { .. } => {
size_summary.actions_accounting_unknown(retained);
}
SizeResolution::Corrupt { .. } => {}
}
}
}
}
IlmAction::DeleteVersionAction => {
if let Some(opt) = object_opts.get(i) {
to_delete_objs.push(ObjectToDelete {
object_name: opt.name.clone(),
version_id: opt.version_id,
..Default::default()
});
noncurrent_events.push(event.clone());
noncurrent_unknown.push(oi);
}
}
IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
let trace_started_at = trace_start_instant();
let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
let done_ilm = Metrics::time_ilm(event.action);
done_ilm(1)();
}
size_summary.actions_accounting_unknown(oi);
}
IlmAction::NoneAction | IlmAction::ActionCount => {
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
}
size_summary.actions_accounting_unknown(oi);
}
}
continue;
} }
}; };
@@ -446,36 +801,24 @@ impl ScannerItem {
done_ilm(1)(); done_ilm(1)();
remaining_versions = 0; remaining_versions = 0;
} else { } else {
PendingScannerAccounting { if let Some(actual_size) = known_size {
object: oi,
retained_size: actual_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
for retained in object_infos.iter().skip(i + 1) {
let retained_size = match retained.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %retained.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
}
};
PendingScannerAccounting { PendingScannerAccounting {
object: retained, object: oi,
retained_size, retained_size: actual_size,
expired_size: 0, expired_size: 0,
} }
.apply(size_summary, &mut cumulative_size, false); .apply(size_summary, &mut cumulative_size, false);
} }
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
if let Some(retained_size) = resolved_sizes[j].known_size() {
PendingScannerAccounting {
object: retained,
retained_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
}
}
} }
break 'eventLoop; break 'eventLoop;
} }
@@ -511,11 +854,13 @@ impl ScannerItem {
version_id: opt.version_id, version_id: opt.version_id,
..Default::default() ..Default::default()
}); });
noncurrent_accounting.push(PendingScannerAccounting { if let Some(actual_size) = known_size {
object: oi, noncurrent_accounting.push(PendingScannerAccounting {
retained_size: actual_size, object: oi,
expired_size: 0, retained_size: actual_size,
}); expired_size: 0,
});
}
account_now = false; account_now = false;
} }
noncurrent_events.push(event.clone()); noncurrent_events.push(event.clone());
@@ -548,7 +893,7 @@ impl ScannerItem {
if account_now { if account_now {
size_summary.actions_accounting(oi, size, actual_size); size_summary.actions_accounting(oi, size, actual_size);
cumulative_size += size; cumulative_size = cumulative_size.saturating_add(size);
} }
} }
} }
@@ -576,11 +921,20 @@ impl ScannerItem {
} }
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) { if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
done_ilm(count)(); done_ilm(count)();
remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len()); remaining_versions = remaining_versions_after_queued_noncurrent(
remaining_versions,
noncurrent_accounting.len(),
noncurrent_unknown.len(),
);
} }
for pending in noncurrent_accounting { for pending in noncurrent_accounting {
pending.apply(size_summary, &mut cumulative_size, queued); pending.apply(size_summary, &mut cumulative_size, queued);
} }
if !queued {
for object in noncurrent_unknown {
size_summary.actions_accounting_unknown(object);
}
}
} }
self.alert_excessive_versions(remaining_versions, cumulative_size); self.alert_excessive_versions(remaining_versions, cumulative_size);
} }
@@ -929,4 +1283,394 @@ mod tests {
assert_eq!(item.object_name, "object"); assert_eq!(item.object_name, "object");
assert_eq!(item.object_path(), "object"); assert_eq!(item.object_path(), "object");
} }
#[test]
fn size_resolution_rejects_negative_overflow_and_unknown_compression() {
let compressed = |actual_size: i64, declared: Option<&str>| {
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
if let Some(declared) = declared {
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, declared.to_string());
}
ObjectInfo {
size: 12,
actual_size,
user_defined: Arc::new(user_defined),
..Default::default()
}
};
let normal = ObjectInfo {
size: 12,
actual_size: 10,
..Default::default()
};
assert_eq!(
resolve_size(&normal),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
let stale_declared_metadata = ObjectInfo {
size: 12,
actual_size: 10,
user_defined: Arc::new(HashMap::from([("x-rustfs-internal-actual-size".to_string(), "not-a-size".to_string())])),
parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
actual_size: -2,
..Default::default()
}]),
..Default::default()
};
assert_eq!(
resolve_size(&stale_declared_metadata),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(0, Some("9"))),
SizeResolution::Known {
logical: 9,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(-1, None)),
SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
}
);
assert!(matches!(
resolve_size(&compressed(0, Some("not-a-size"))),
SizeResolution::Corrupt {
reason: SizeResolutionReason::InvalidDeclaredSize,
..
}
));
assert!(matches!(
resolve_size(&ObjectInfo {
size: 12,
actual_size: -2,
..Default::default()
}),
SizeResolution::Corrupt { .. }
));
assert!(matches!(resolve_size(&compressed(0, Some("-1"))), SizeResolution::Corrupt { .. }));
assert!(matches!(resolve_size(&compressed(0, Some(""))), SizeResolution::Corrupt { .. }));
let unsupported = {
let mut object = compressed(0, None);
let mut metadata = (*object.user_defined).clone();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "unsupported".to_string());
object.user_defined = Arc::new(metadata);
object
};
assert!(matches!(resolve_size(&unsupported), SizeResolution::Corrupt { .. }));
let invalid_part = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
size: 12,
actual_size: -2,
..Default::default()
}]);
object
};
assert!(matches!(resolve_size(&invalid_part), SizeResolution::Corrupt { .. }));
let overflow = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: i64::MAX,
..Default::default()
},
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: 1,
..Default::default()
},
]);
object
};
assert!(matches!(resolve_size(&overflow), SizeResolution::Corrupt { .. }));
let mismatch = compressed(0, None);
assert!(matches!(resolve_size(&mismatch), SizeResolution::Corrupt { .. }));
assert_eq!(
resolve_size(&ObjectInfo {
size: 0,
actual_size: 0,
..Default::default()
}),
SizeResolution::Known { logical: 0, physical: 0 }
);
}
#[test]
fn size_resolution_records_and_replays_one_identity() {
let version_id = uuid::Uuid::new_v4();
let generation = uuid::Uuid::new_v4();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "not-a-number".to_string());
let corrupt = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
version_id: Some(version_id),
data_dir: Some(generation),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
let resolution = resolve_size(&corrupt);
record_size_resolution(&mut summary, &corrupt, &resolution);
record_size_resolution(&mut summary, &corrupt, &resolution);
assert_eq!(summary.size_reconciliation.len(), 1);
assert_eq!(summary.size_reconciliation[0].reason, "invalid_declared_size");
assert_eq!(summary.size_reconciliation[0].physical_size, Some(12));
let known = ObjectInfo {
actual_size: 12,
user_defined: Arc::new(HashMap::new()),
..corrupt.clone()
};
record_size_resolution(&mut summary, &known, &resolve_size(&known));
summary.record_reconciliation_scope(&known.bucket, &known.name);
assert_eq!(summary.reconciliation_scopes.len(), 1);
assert_eq!(summary.reconciliation_scopes[0].bucket, "bucket");
}
#[test]
fn malformed_size_has_same_ilm_accounting() {
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "invalid".to_string());
let object = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
user_defined: Arc::new(metadata),
..Default::default()
};
let resolution = resolve_size(&object);
let mut without_ilm = SizeSummary::default();
let mut with_ilm = SizeSummary::default();
record_size_resolution(&mut without_ilm, &object, &resolution);
record_size_resolution(&mut with_ilm, &object, &resolution);
assert_eq!(without_ilm.size_reconciliation, with_ilm.size_reconciliation);
assert_eq!(without_ilm.total_size, 0);
assert_eq!(with_ilm.total_size, 0);
assert!(without_ilm.tier_stats.is_empty());
assert!(with_ilm.tier_stats.is_empty());
}
#[test]
fn size_resolution_parses_once_per_version() {
let objects = vec![
ObjectInfo {
bucket: "bucket".to_string(),
name: "one".to_string(),
size: 1,
actual_size: 1,
..Default::default()
},
ObjectInfo {
bucket: "bucket".to_string(),
name: "two".to_string(),
size: 2,
actual_size: -2,
..Default::default()
},
];
let resolutions = resolve_sizes(&objects);
assert_eq!(resolutions.len(), objects.len());
assert!(matches!(resolutions[0], SizeResolution::Known { logical: 1, .. }));
assert!(matches!(resolutions[1], SizeResolution::Corrupt { .. }));
}
#[test]
fn queued_unknown_noncurrent_versions_are_removed_from_alert_count() {
assert_eq!(remaining_versions_after_queued_noncurrent(3, 1, 2), 0);
assert_eq!(remaining_versions_after_queued_noncurrent(7, 2, 1), 4);
assert_eq!(remaining_versions_after_queued_noncurrent(usize::MAX, usize::MAX, usize::MAX), 0);
}
#[test]
fn malformed_size_blocks_size_dependent_transition_but_allows_time_only_expiry() {
let size_filtered = BucketLifecycleConfiguration {
rules: vec![s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("size".to_string()),
filter: Some(s3s::dto::LifecycleRuleFilter {
object_size_greater_than: Some(1),
..Default::default()
}),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
..Default::default()
};
let unknown = SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
let size_event = Event {
action: IlmAction::DeleteAction,
rule_id: "size".to_string(),
..Default::default()
};
assert!(!lifecycle_event_allowed(&unknown, &size_event, &size_filtered));
assert!(!lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
rule_id: "size".to_string(),
..Default::default()
},
&size_filtered
));
let mixed_filters = BucketLifecycleConfiguration {
rules: vec![
size_filtered.rules[0].clone(),
s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("time".to_string()),
filter: None,
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
},
],
..Default::default()
};
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time".to_string(),
..Default::default()
},
&mixed_filters
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&SizeResolution::Corrupt {
physical: 12,
reason: SizeResolutionReason::InvalidDeclaredSize,
},
&Event {
action: IlmAction::DeleteAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(!lifecycle_event_allowed(
&SizeResolution::Corrupt {
physical: 12,
reason: SizeResolutionReason::InvalidDeclaredSize,
},
&Event {
action: IlmAction::DeleteAction,
rule_id: "size".to_string(),
..Default::default()
},
&size_filtered
));
assert!(lifecycle_rule_has_size_filter(
&BucketLifecycleConfiguration {
rules: vec![s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: None,
filter: Some(s3s::dto::LifecycleRuleFilter {
object_size_greater_than: Some(1),
..Default::default()
}),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
..Default::default()
},
""
));
assert!(lifecycle_event_allowed(
&SizeResolution::Known {
logical: 10,
physical: 12,
},
&Event {
action: IlmAction::DeleteAllVersionsAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time-only".to_string(),
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
}
#[tokio::test]
async fn long_object_size_reconciliation_scope_uses_bounded_identity() {
let object_name = "o".repeat(600);
let mut item = scanner_item_with_prefix("");
item.object_name = object_name.clone();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
let object = ObjectInfo {
bucket: item.bucket.clone(),
name: object_name.clone(),
size: 12,
actual_size: -1,
version_id: Some(uuid::Uuid::new_v4()),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
item.apply_actions(vec![object], None, VersioningConfiguration::default(), &mut summary)
.await;
let bounded_bucket = bounded_reconciliation_field(&item.bucket);
let bounded_object = bounded_reconciliation_field(&object_name);
assert_eq!(summary.reconciliation_scopes[0].bucket, bounded_bucket);
assert_eq!(summary.reconciliation_scopes[0].object, bounded_object);
assert_eq!(summary.size_reconciliation[0].object, bounded_object);
assert_eq!(summary.versions, 1);
assert_eq!(summary.total_size, 0);
}
} }
@@ -326,6 +326,9 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
skip_heal: Arc::new(AtomicBool::new(false)), skip_heal: Arc::new(AtomicBool::new(false)),
local_disk: disk, local_disk: disk,
pending_heals_changed: false, pending_heals_changed: false,
pending_size_reconciliation_keys: HashSet::new(),
pending_size_reconciliation_scopes: HashSet::new(),
pending_size_reconciliation_truncated: false,
list_path_raw_options_observer: None, list_path_raw_options_observer: None,
}; };
@@ -388,6 +391,66 @@ async fn test_record_failed_ttl_zero_noop() {
assert!(!scanner.should_skip_failed("path2")); assert!(!scanner.should_skip_failed("path2"));
} }
#[tokio::test]
async fn malformed_size_reconciliation_replays_after_restart() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let entry = SizeReconciliationEntry {
key: "1:b|6:object|0:|0:".to_string(),
bucket: "b".to_string(),
object: "object".to_string(),
reason: "invalid_declared_size".to_string(),
physical_size: Some(12),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry.clone());
summary.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&summary);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.update_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.new_cache.info.size_reconciliation[&entry.key].attempts, 2);
let encoded = rmp_serde::to_vec_named(&scanner.new_cache.info).expect("size ledger should encode");
let decoded: crate::data_usage_define::DataUsageCacheInfo =
rmp_serde::from_slice(&encoded).expect("size ledger should decode");
assert_eq!(decoded.size_reconciliation.len(), 1);
assert_eq!(decoded.size_reconciliation[&entry.key].reason, "invalid_declared_size");
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
assert!(scanner.update_cache.info.size_reconciliation.is_empty());
}
#[tokio::test]
async fn malformed_size_reconciliation_clears_bounded_long_object_scope() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let long_object = "o".repeat(600);
let bounded_object = item_actions::bounded_reconciliation_field(&long_object);
let entry = SizeReconciliationEntry {
key: "long-object-key".to_string(),
bucket: "b".to_string(),
object: bounded_object,
reason: "invalid_declared_size".to_string(),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", &long_object);
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
}
#[test] #[test]
fn test_classify_get_size_failure_marks_metadata_heal_object_path() { fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
let temp_dir = std::env::temp_dir(); let temp_dir = std::env::temp_dir();
+12
View File
@@ -245,6 +245,18 @@ impl TestECStoreEnvBuilder {
.await .await
.expect("build test ECStore"); .expect("build test ECStore");
// The production bootstrap only persists pool.bin from the elected
// first cluster node. Test stores intentionally have no cluster
// election, but heal-format still requires that durable fence before
// it can write any disk format. Materialize the validated topology
// here so the shared fixture models a ready single-node store.
let mut pool_meta = ecstore.pool_meta.read().await.clone();
pool_meta.dont_save = false;
pool_meta
.save(ecstore.pools.clone())
.await
.expect("persist test pool metadata");
if self.init_bucket_metadata { if self.init_bucket_metadata {
let buckets_list = ecstore let buckets_list = ecstore
.list_bucket(&BucketOptions { .list_bucket(&BucketOptions {
-3
View File
@@ -43,9 +43,6 @@ allow-git = [
# RustFS fork carrying presigned expiry and constant-time authentication fixes. # RustFS fork carrying presigned expiry and constant-time authentication fixes.
# owner: rustfs-maintainers review: 2026-10 # owner: rustfs-maintainers review: 2026-10
"https://github.com/rustfs/s3s.git", "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] [bans]
+149
View File
@@ -0,0 +1,149 @@
# 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.
+4 -4
View File
@@ -322,7 +322,7 @@ thiserror = { workspace = true }
tracing.workspace = true tracing.workspace = true
url = { workspace = true } url = { workspace = true }
urlencoding = { workspace = true } urlencoding = { workspace = true }
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
zip = { workspace = true } zip = { workspace = true }
libc = { workspace = true } libc = { workspace = true }
rand = { workspace = true, features = ["serde"] } rand = { workspace = true, features = ["serde"] }
@@ -336,16 +336,16 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true } tracing-opentelemetry = { workspace = true }
# Data structures # Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] } hashbrown = { workspace = true, features = ["serde", "rayon"] }
mimalloc = { workspace = true } rustfs-mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies] [target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies] [target.'cfg(not(target_os = "windows"))'.dependencies]
libmimalloc-sys.workspace = true rustfs-mimalloc-sys.workspace = true
[dev-dependencies] [dev-dependencies]
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
serial_test = { workspace = true } serial_test = { workspace = true }
tempfile = { workspace = true } tempfile = { workspace = true }
aws-config = { workspace = true } aws-config = { workspace = true }
+121 -32
View File
@@ -41,7 +41,7 @@ use crate::admin::storage_api::config::save_admin_config;
use crate::admin::storage_api::contract::bucket::{ use crate::admin::storage_api::contract::bucket::{
BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp, BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp,
}; };
use crate::admin::storage_api::error::Error as StorageError; use crate::admin::storage_api::error::{Error as StorageError, is_err_bucket_not_found};
use crate::admin::storage_api::runtime::ECStore; use crate::admin::storage_api::runtime::ECStore;
use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body}; use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body};
use crate::auth::constant_time_eq; use crate::auth::constant_time_eq;
@@ -55,6 +55,7 @@ use crate::storage::storage_api::{
use base64::Engine; use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::engine::general_purpose::URL_SAFE_NO_PAD; use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use futures::StreamExt;
use hmac::{Hmac, Mac}; use hmac::{Hmac, Mac};
use http::header::{CONTENT_TYPE, HOST}; use http::header::{CONTENT_TYPE, HOST};
use http::{HeaderMap, HeaderValue, Uri}; use http::{HeaderMap, HeaderValue, Uri};
@@ -2096,6 +2097,18 @@ async fn remote_add_preflight_info(site: &PeerSite) -> S3Result<SiteReplicationA
format!("invalid site replication metainfo from `{}`: {e}", site.endpoint), format!("invalid site replication metainfo from `{}`: {e}", site.endpoint),
) )
})?; })?;
if info.deployment_id.is_empty() {
// The peer will be tracked under a locally derived fallback ID
// (deployment_id_for_endpoint) instead of its real deployment ID.
warn!(
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
result = "peer_deployment_id_missing",
peer_endpoint = %site.endpoint,
"admin site replication state"
);
}
let idp_body = send_peer_admin_get_request_with_client( let idp_body = send_peer_admin_get_request_with_client(
&client, &client,
@@ -2206,20 +2219,30 @@ fn site_replication_bootstrap_token(uri: &Uri) -> Option<String> {
query_pairs(uri).get("bootstrapToken").cloned() query_pairs(uri).get("bootstrapToken").cloned()
} }
fn bootstrap_bucket_make_op_path(bucket: &SRBucketInfo) -> String { /// Query for a peer `make-with-versioning` bucket op. `versioningEnabled`
/// always travels so the outbound query matches MinIO's site-replication
/// make-bucket wire contract: MinIO's own create-bucket hook sends
/// `versioningEnabled=true` on this op. RustFS's inbound handler
/// force-enables versioning either way.
fn make_with_versioning_bucket_op_path(bucket: &str, created_at: Option<&str>, lock_enabled: bool) -> String {
let mut query = form_urlencoded::Serializer::new(String::new()); let mut query = form_urlencoded::Serializer::new(String::new());
query.append_pair("bucket", &bucket.bucket); query.append_pair("bucket", bucket);
query.append_pair("operation", "make-with-versioning"); query.append_pair("operation", SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING);
if let Some(created_at) = bucket query.append_pair("versioningEnabled", "true");
.created_at if let Some(created_at) = created_at {
.and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok()) query.append_pair("createdAt", created_at);
{
query.append_pair("createdAt", &created_at);
} }
if bucket.object_lock_config.is_some() { if lock_enabled {
query.append_pair("lockEnabled", "true"); query.append_pair("lockEnabled", "true");
} }
format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?{}", query.finish()) format!("{SITE_REPLICATION_PEER_BUCKET_OPS_PATH}?{}", query.finish())
}
fn bootstrap_bucket_make_op_path(bucket: &SRBucketInfo) -> String {
let created_at = bucket
.created_at
.and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok());
make_with_versioning_bucket_op_path(&bucket.bucket, created_at.as_deref(), bucket.object_lock_config.is_some())
} }
fn bootstrap_bucket_meta_item(bucket: &SRBucketInfo, item_type: &str, updated_at: Option<OffsetDateTime>) -> SRBucketMeta { fn bootstrap_bucket_meta_item(bucket: &SRBucketInfo, item_type: &str, updated_at: Option<OffsetDateTime>) -> SRBucketMeta {
@@ -4246,16 +4269,7 @@ async fn broadcast_site_replication_make_bucket(
.format(&time::format_description::well_known::Rfc3339) .format(&time::format_description::well_known::Rfc3339)
.unwrap_or_default(); .unwrap_or_default();
let path = { let path = make_with_versioning_bucket_op_path(bucket, Some(&created_at), lock_enabled);
let mut query = form_urlencoded::Serializer::new(String::new());
query.append_pair("bucket", bucket);
query.append_pair("operation", "make-with-versioning");
query.append_pair("createdAt", &created_at);
if lock_enabled {
query.append_pair("lockEnabled", "true");
}
format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?{}", query.finish())
};
let path = if let Some(token) = bootstrap_token { let path = if let Some(token) = bootstrap_token {
with_site_replication_bootstrap_token(&path, token) with_site_replication_bootstrap_token(&path, token)
} else { } else {
@@ -10206,13 +10220,25 @@ impl Operation for SiteReplicationStatusHandler {
} }
} }
/// `POST /v3/site-replication/devnull` — peer link-check upload drain.
/// MinIO streams multi-megabyte probe bodies here during site netperf link
/// checks and expects an unbounded discard (its handler copies to io.Discard);
/// buffering through the 1MB admin body cap turned any larger probe into a
/// 400 and a false link failure. Stream and discard instead — no size cap.
async fn drain_site_replication_devnull(mut input: Body) -> S3Result<()> {
while let Some(chunk) = input.next().await {
chunk.map_err(|e| s3_error!(InvalidRequest, "failed to read devnull stream: {}", e))?;
}
Ok(())
}
pub struct SiteReplicationDevNullHandler {} pub struct SiteReplicationDevNullHandler {}
#[async_trait::async_trait] #[async_trait::async_trait]
impl Operation for SiteReplicationDevNullHandler { impl Operation for SiteReplicationDevNullHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> { async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?;
let _ = read_plain_admin_body(req.input).await?; drain_site_replication_devnull(req.input).await?;
Ok(empty_response(StatusCode::NO_CONTENT)) Ok(empty_response(StatusCode::NO_CONTENT))
} }
} }
@@ -10471,6 +10497,19 @@ impl Operation for SRPeerJoinHandler {
} }
} }
/// Outcome of a peer-driven `purge-deleted-bucket` replay. A bucket that is
/// already gone means the purge raced an earlier replay or a local delete —
/// that is success — but any other failure must reach the sender like the
/// sibling delete branches do: swallowing it answered 200 while the bucket
/// survived on this site.
fn purge_deleted_bucket_result(result: Result<(), StorageError>) -> S3Result<()> {
match result {
Ok(()) => Ok(()),
Err(err) if is_err_bucket_not_found(&err) => Ok(()),
Err(err) => Err(ApiError::from(err).into()),
}
}
pub struct SRPeerBucketOpsHandler {} pub struct SRPeerBucketOpsHandler {}
#[async_trait::async_trait] #[async_trait::async_trait]
@@ -10570,16 +10609,18 @@ impl Operation for SRPeerBucketOpsHandler {
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
} }
"purge-deleted-bucket" => { "purge-deleted-bucket" => {
let _ = store purge_deleted_bucket_result(
.delete_bucket( store
&bucket, .delete_bucket(
&DeleteBucketOptions { &bucket,
force: true, &DeleteBucketOptions {
srdelete_op: SRBucketDeleteOp::Purge, force: true,
..Default::default() srdelete_op: SRBucketDeleteOp::Purge,
}, ..Default::default()
) },
.await; )
.await,
)?;
} }
_ => return Err(s3_error!(InvalidRequest, "unsupported site replication bucket operation")), _ => return Err(s3_error!(InvalidRequest, "unsupported site replication bucket operation")),
} }
@@ -13925,6 +13966,54 @@ mod tests {
assert!(!query_flag(&uri, "missing")); assert!(!query_flag(&uri, "missing"));
} }
/// A5 red-light: a `purge-deleted-bucket` replay must report success when
/// the bucket is already gone, and must propagate every other failure —
/// the swallowed error answered 200 while the bucket survived.
#[test]
fn test_purge_deleted_bucket_result_tolerates_only_missing_bucket() {
assert!(purge_deleted_bucket_result(Ok(())).is_ok());
assert!(purge_deleted_bucket_result(Err(StorageError::BucketNotFound("photos".to_string()))).is_ok());
assert!(purge_deleted_bucket_result(Err(StorageError::VolumeNotFound)).is_ok());
let err = purge_deleted_bucket_result(Err(StorageError::StorageFull))
.expect_err("non-not-found delete failures must propagate");
assert_ne!(*err.code(), S3ErrorCode::NoSuchBucket);
}
/// C5 red-light: the site-replication devnull drain must accept bodies
/// beyond the 1MB admin body cap — MinIO's link check streams large
/// probe bodies and treats a 400 as a broken link.
#[tokio::test]
async fn test_site_replication_devnull_drains_body_beyond_admin_cap() {
let body = Body::from(vec![0u8; MAX_ADMIN_REQUEST_BODY_SIZE + 1]);
drain_site_replication_devnull(body)
.await
.expect("devnull must drain bodies larger than the admin body cap");
}
/// A3 red-light: `versioningEnabled` must travel on every outbound
/// make-with-versioning bucket op so the query matches MinIO's
/// site-replication make-bucket wire contract (MinIO's own hook sends
/// `versioningEnabled=true` on this op).
#[test]
fn test_make_with_versioning_op_paths_send_versioning_enabled() {
let bucket = SRBucketInfo {
bucket: "photos".to_string(),
created_at: Some(OffsetDateTime::UNIX_EPOCH),
object_lock_config: Some(BASE64_STANDARD.encode("<ObjectLockConfiguration/>")),
..Default::default()
};
let bootstrap = bootstrap_bucket_make_op_path(&bucket);
assert!(bootstrap.contains("operation=make-with-versioning"), "{bootstrap}");
assert!(bootstrap.contains("versioningEnabled=true"), "{bootstrap}");
assert!(bootstrap.contains("createdAt="), "{bootstrap}");
assert!(bootstrap.contains("lockEnabled=true"), "{bootstrap}");
// The broadcast path (create-bucket hook) shares the same builder.
let broadcast = make_with_versioning_bucket_op_path("photos", Some("1970-01-01T00:00:00Z"), false);
assert!(broadcast.contains("versioningEnabled=true"), "{broadcast}");
assert!(!broadcast.contains("lockEnabled"), "{broadcast}");
}
#[tokio::test] #[tokio::test]
#[serial] #[serial]
async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() { async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() {
+24 -5
View File
@@ -13,9 +13,9 @@
// limitations under the License. // limitations under the License.
use rustfs_madmin::{PeerInfo, SyncStatus}; use rustfs_madmin::{PeerInfo, SyncStatus};
use std::collections::{BTreeMap, hash_map::DefaultHasher}; use std::collections::BTreeMap;
use std::hash::{Hash, Hasher};
use url::Url; use url::Url;
use uuid::Uuid;
fn has_http_scheme(endpoint: &str) -> bool { fn has_http_scheme(endpoint: &str) -> bool {
endpoint.get(..7).is_some_and(|prefix| prefix.eq_ignore_ascii_case("http://")) endpoint.get(..7).is_some_and(|prefix| prefix.eq_ignore_ascii_case("http://"))
@@ -66,10 +66,12 @@ pub fn site_identity_key(endpoint: &str) -> String {
.unwrap_or_else(|| trimmed.to_ascii_lowercase()) .unwrap_or_else(|| trimmed.to_ascii_lowercase())
} }
/// Fallback deployment ID for a peer that reported none. UUIDv5 over the
/// canonical endpoint: the ID is persisted in site-replication state and
/// broadcast to peers, so it must be identical across Rust toolchains
/// (`DefaultHasher` is not) and across spellings of the same endpoint.
pub fn deployment_id_for_endpoint(endpoint: &str) -> String { pub fn deployment_id_for_endpoint(endpoint: &str) -> String {
let mut hasher = DefaultHasher::new(); Uuid::new_v5(&Uuid::NAMESPACE_URL, canonical_endpoint(endpoint).as_bytes()).to_string()
endpoint.hash(&mut hasher);
format!("{:016x}", hasher.finish())
} }
pub fn same_identity_endpoint(left: &str, right: &str) -> bool { pub fn same_identity_endpoint(left: &str, right: &str) -> bool {
@@ -174,6 +176,23 @@ mod tests {
} }
} }
/// B8 red-light: the fallback deployment ID must be a toolchain-stable
/// UUIDv5 over the canonical endpoint — `DefaultHasher` output is not
/// guaranteed stable across Rust releases, yet the ID is persisted in
/// site-replication state and broadcast to peers.
#[test]
fn deployment_id_for_endpoint_is_stable_uuid_v5_over_canonical_endpoint() {
let endpoint = "https://node-a.example.com:9000";
let id = deployment_id_for_endpoint(endpoint);
let parsed = uuid::Uuid::parse_str(&id).expect("fallback deployment ID must be a UUID");
assert_eq!(parsed.get_version_num(), 5, "fallback deployment ID must be UUIDv5");
// Deterministic for the same endpoint and for spelling variants that
// share a canonical form; distinct endpoints stay distinct.
assert_eq!(id, deployment_id_for_endpoint(endpoint));
assert_eq!(id, deployment_id_for_endpoint(" HTTPS://Node-A.Example.Com:9000/ "));
assert_ne!(id, deployment_id_for_endpoint("https://node-b.example.com:9000"));
}
#[test] #[test]
fn canonical_endpoint_accepts_case_insensitive_scheme() { fn canonical_endpoint_accepts_case_insensitive_scheme() {
assert_eq!( assert_eq!(
+2 -1
View File
@@ -51,7 +51,7 @@ mod ecstore_disk {
} }
mod ecstore_error { mod ecstore_error {
pub(crate) use crate::storage::storage_api::ecstore_error::StorageError; pub(crate) use crate::storage::storage_api::ecstore_error::{StorageError, is_err_bucket_not_found};
} }
#[allow(unused_imports)] #[allow(unused_imports)]
@@ -919,6 +919,7 @@ pub(crate) mod contract {
} }
pub(crate) mod error { pub(crate) mod error {
pub(crate) use super::ecstore_error::is_err_bucket_not_found;
pub(crate) use super::{Error, StorageError}; pub(crate) use super::{Error, StorageError};
} }
+1 -7
View File
@@ -369,14 +369,8 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
} }
#[cfg(not(target_os = "windows"))] #[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> { fn collect_allocator_memory(force: bool) -> Result<(), String> {
// SAFETY: `mi_collect` is provided by the active global allocator backend rustfs_mimalloc::MiMalloc::collect(force);
// 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(()) Ok(())
} }
+10 -8
View File
@@ -26,22 +26,22 @@ struct MiMallocAllocator;
unsafe impl GlobalAlloc for MiMallocAllocator { unsafe impl GlobalAlloc for MiMallocAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 { unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout. // SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { mimalloc::MiMalloc.alloc(layout) } unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
} }
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout. // SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) } unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
} }
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged. // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) } unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
} }
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { 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. // SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) } unsafe { rustfs_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")))] #[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator] #[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc; static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
fn main() { fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build(); let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,8 +71,9 @@ mod tests {
allocation.extend_from_slice(&[7_u8; 64]); allocation.extend_from_slice(&[7_u8; 64]);
assert_eq!(allocation.len(), 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. // SAFETY: the live Vec pointer is valid to inspect for heap ownership.
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) }); assert!(unsafe { heap.contains(allocation.as_ptr()) });
} }
#[test] #[test]
@@ -85,12 +86,13 @@ mod tests {
let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout"); 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 grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
let allocator = super::MiMallocAllocator; let allocator = super::MiMallocAllocator;
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: The pointer is checked for null before use and later released // SAFETY: The pointer is checked for null before use and later released
// through the same allocator with the corresponding layout. // through the same allocator with the corresponding layout.
let ptr = unsafe { allocator.alloc_zeroed(layout) }; let ptr = unsafe { allocator.alloc_zeroed(layout) };
assert!(!ptr.is_null()); assert!(!ptr.is_null());
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) }); assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) }); assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure // SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -102,7 +104,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test"); panic!("mimalloc realloc failed in allocator smoke test");
} }
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) }); assert!(unsafe { heap.contains(grown_ptr) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released // SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout. // with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) }; unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+19 -36
View File
@@ -17,10 +17,7 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split, record_cpu_usage, record_memory_usage, record_process_memory_split,
}; };
use serde::Serialize; use serde::Serialize;
#[cfg(any(test, not(target_os = "windows")))]
use serde_json::Value; use serde_json::Value;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path; use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock}; use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration; use std::time::Duration;
@@ -231,7 +228,18 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1) read_cgroup_v2().or_else(read_cgroup_v1)
} }
#[cfg(any(test, not(target_os = "windows")))] 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,
})
}
fn numeric_json_value(value: &Value) -> Option<u64> { fn numeric_json_value(value: &Value) -> Option<u64> {
match value { match value {
Value::Number(number) => number Value::Number(number) => number
@@ -242,7 +250,6 @@ 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> { fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value { match value {
Value::Object(fields) => fields Value::Object(fields) => fields
@@ -254,7 +261,6 @@ 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> { fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value { match value {
Value::Object(fields) => { Value::Object(fields) => {
@@ -271,12 +277,10 @@ 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> { fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current") 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> { fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
metrics metrics
.iter() .iter()
@@ -285,7 +289,6 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
.filter(|value| *value > 0) .filter(|value| *value > 0)
} }
#[cfg(any(test, not(target_os = "windows")))]
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> { fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?; let value = serde_json::from_str::<Value>(stats_json).ok()?;
let malloc_metrics = ["malloc_normal", "malloc_huge"]; let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -312,33 +315,6 @@ 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 { fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1) rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
} }
@@ -566,6 +542,13 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None); 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] #[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() { fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false); let snapshot = build_memory_observability_status_snapshot(false, 15, false);
+1 -1
View File
@@ -241,7 +241,7 @@ env \
RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \ RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \ RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
cargo test -p rustfs-kms --test vault_ha_failover_live \ cargo test -p rustfs-kms --test vault_ha_failover_live \
vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \ vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
--ignored --nocapture --test-threads=1 & --ignored --nocapture --test-threads=1 &
TEST_PID=$! TEST_PID=$!