Compare commits

..

1 Commits

Author SHA1 Message Date
overtrue 432a56f6e1 ci: pin s3tests Python tools 2026-08-23 03:19:26 +08:00
29 changed files with 249 additions and 1714 deletions
+6 -1
View File
@@ -178,9 +178,14 @@ jobs:
- name: Install Python tools
run: |
python3 -m pip install --user --upgrade pip awscurl tox
python3 -m pip install --user --upgrade pip "awscurl==0.44" "tox==4.60.0"
echo "$HOME/.local/bin" >> "$GITHUB_PATH"
- name: Verify Python tools
run: |
test "$(python3 -c 'import importlib.metadata as m; print(m.version("awscurl"))')" = "0.44"
test "$(python3 -c 'import importlib.metadata as m; print(m.version("tox"))')" = "4.60.0"
- name: Enable buildx
uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3
+1 -2
View File
@@ -30,8 +30,7 @@ make build-docker BUILD_OS=ubuntu22.04
- Crate membership: `Cargo.toml` `[workspace].members`
- Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
- Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
-2
View File
@@ -70,8 +70,6 @@ make pre-pr
> For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
### 🔒 Automated Pre-commit Hooks
#### What `make pre-commit` and `make pre-pr` actually run
Generated
+27 -23
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.4"
version = "1.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,6 +2522,12 @@ dependencies = [
"subtle",
]
[[package]]
name = "cty"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
@@ -5982,6 +5988,15 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
[[package]]
name = "libmimalloc-sys"
version = "0.1.49"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"cc",
"cty",
]
[[package]]
name = "libredox"
version = "0.1.20"
@@ -6382,6 +6397,14 @@ dependencies = [
"synstructure 0.13.2",
]
[[package]]
name = "mimalloc"
version = "0.1.52"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"libmimalloc-sys",
]
[[package]]
name = "mime"
version = "0.3.17"
@@ -9139,11 +9162,13 @@ dependencies = [
"insta",
"jiff",
"libc",
"libmimalloc-sys",
"libsystemd",
"matchit 0.9.2",
"md-5 0.11.0",
"metrics",
"metrics-util",
"mimalloc",
"mime_guess",
"opentelemetry",
"opentelemetry_sdk",
@@ -9179,8 +9204,6 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9852,24 +9875,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "rustfs-mimalloc"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
dependencies = [
"rustfs-mimalloc-sys",
]
[[package]]
name = "rustfs-mimalloc-sys"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
dependencies = [
"cc",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.3"
@@ -12683,7 +12688,6 @@ dependencies = [
"js-sys",
"rand 0.10.2",
"serde_core",
"sha1_smol",
"wasm-bindgen",
]
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+1 -1
View File
@@ -2026,7 +2026,7 @@ impl PoolMeta {
self.load_no_lock(pool).await
}
pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
where
S: EcstoreObjectIO,
{
+8 -20
View File
@@ -988,11 +988,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for Sets {
}
}
impl Sets {
pub(crate) async fn heal_format_with_fence<F>(&self, dry_run: bool, fence_lost: F) -> Result<(HealResultItem, Option<Error>)>
where
F: Fn() -> bool + Send + Sync,
{
#[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>)> {
let (disks, init_errs) = init_storage_disks_with_errors(
&self.endpoints.endpoints,
&DiskOption {
@@ -1065,9 +1068,6 @@ impl Sets {
// Save new formats `format.json` on unformatted disks.
for (index, (fm, disk)) in tmp_new_formats.iter_mut().zip(disks.iter()).enumerate() {
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 Some(disk) = disk.as_ref() {
let _ = disk.close().await;
@@ -1101,18 +1101,6 @@ impl Sets {
}
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))]
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
let mut result = HealResultItem {
+6 -38
View File
@@ -784,24 +784,6 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
///
/// # Returns
/// A Result containing the BitrotWriterWrapper or an error
/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
///
/// A known length is grown by one checksum per shard so the on-disk file size
/// matches what the bitrot writer emits. A negative length is the
/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
/// a fixed body length when locating the authenticated trailer. Clamping it
/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
/// because a genuinely empty object still means an empty body.
fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
if length <= 0 {
return length;
}
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
}
pub async fn create_bitrot_writer(
is_inline_buffer: bool,
disk: Option<&DiskStore>,
@@ -814,7 +796,12 @@ pub async fn create_bitrot_writer(
let writer = if is_inline_buffer {
CustomWriter::new_inline_buffer()
} else if let Some(disk) = disk {
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
let length = if length > 0 {
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
} else {
0
};
let file = disk.create_file("", volume, path, length).await?;
#[cfg(feature = "hotpath")]
@@ -833,25 +820,6 @@ mod tests {
use rustfs_rio::ChunkReader;
use std::collections::VecDeque;
#[test]
fn bitrot_create_file_size_grows_known_length_by_checksums() {
// 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
}
#[test]
fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
// SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
// put_file_stream receiver relies on a non-positive size to parse the auth
// trailer from the stream tail, so the sentinel must survive untouched.
assert_eq!(
bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
);
}
struct TestChunkReader {
chunks: VecDeque<Bytes>,
}
+14 -1
View File
@@ -2124,13 +2124,26 @@ impl SetDisks {
let put_object_size = known_put_object_storage_size(data.size());
let shard_file_size_raw = erasure.shard_file_size(put_object_size);
let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let is_inline_buffer =
storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
let shard_file_size = shard_file_size_raw;
let shard_size = erasure.shard_size();
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
{
use std::io::Write;
let msg = format!(
"INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
);
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
let _ = f.write_all(msg.as_bytes());
}
let _ = std::io::stderr().write_all(msg.as_bytes());
}
rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
let (mut writers, errors) = if direct_inline_commit {
+3 -332
View File
@@ -13,12 +13,7 @@
// limitations under the License.
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::namespace::NamespaceLocking as _;
use rustfs_lock::NamespaceLockGuard;
use tracing::trace;
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -35,119 +30,7 @@ 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 {
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>>> {
match opts.pool {
Some(pool_idx) => Ok(vec![
@@ -169,26 +52,9 @@ impl ECStore {
};
let mut count_no_heal = 0;
let mut count_completed = 0;
let mut first_error = None;
for (pool_idx, pool) in self.pools.iter().enumerate() {
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?;
for pool in self.pools.iter() {
let (mut result, err) = pool.heal_format(dry_run).await?;
if let Some(err) = err {
match err {
StorageError::NoHealRequired => {
@@ -203,18 +69,11 @@ impl ECStore {
r.set_count += result.set_count;
r.before.drives.append(&mut result.before.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 {
return Ok((r, Some(err)));
}
if count_no_heal + count_completed == self.pools.len() {
if count_no_heal == self.pools.len() {
info!(
event = EVENT_HEAL_FORMAT_COMPLETED,
component = LOG_COMPONENT_ECSTORE,
@@ -443,7 +302,6 @@ mod tests {
use crate::disk::{DeleteOptions, DiskOption, format::FormatV3, new_disk};
use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use crate::runtime::instance::InstanceContext;
use crate::services::rebalance::{RebalanceInfo, RebalanceStats};
use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions};
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations};
use crate::store::init_format::{load_format_erasure, save_format_file};
@@ -495,164 +353,6 @@ 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) {
let temp_dir = tempfile::tempdir().expect("multi-pool heal test directory should be created");
let mut pool_endpoints = Vec::new();
@@ -1189,18 +889,6 @@ mod tests {
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
.handle_heal_format(false)
.await
@@ -1214,22 +902,5 @@ mod tests {
.await
.expect("the later pool should be healed despite the first pool error");
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
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let accounting = vec![None; objects.len()];
let mut accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
.filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
.collect::<Vec<_>>();
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+1 -632
View File
@@ -21,7 +21,6 @@ use arc_swap::ArcSwapOption;
use rmp::Marker;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::str::from_utf8;
use std::{
fmt::Debug,
@@ -38,13 +37,8 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::spawn;
use tokio::sync::Mutex;
use tracing::{debug, warn};
use uuid::Uuid;
const SLASH_SEPARATOR: &str = "/";
pub const MAX_META_CACHE_HEAL_CANDIDATES: usize = 1024;
/// Keep truncation continuations bounded while still giving the scanner a
/// safe object-level retry for versions that did not fit in the candidate set.
pub const MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS: usize = 64;
#[derive(Clone, Debug, Default)]
pub struct MetadataResolutionParams {
@@ -72,50 +66,6 @@ pub struct MetaCacheEntry {
pub reusable: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum MetaCacheHealCandidateKind {
Object,
DeleteMarker,
UnversionedObject,
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct MetaCacheHealCandidate {
pub object: String,
pub version_id: Option<Uuid>,
pub kind: MetaCacheHealCandidateKind,
/// Number of raw disk entries that carried this validated version.
pub replica_count: usize,
}
impl MetaCacheHealCandidate {
pub fn validated_version(&self) -> Option<Uuid> {
match self.kind {
MetaCacheHealCandidateKind::Object | MetaCacheHealCandidateKind::DeleteMarker => self.version_id,
MetaCacheHealCandidateKind::UnversionedObject => None,
}
}
pub fn is_unversioned(&self) -> bool {
self.kind == MetaCacheHealCandidateKind::UnversionedObject
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct MetaCacheHealDiscovery {
pub candidates: Vec<MetaCacheHealCandidate>,
pub unverified_count: usize,
pub truncated: bool,
/// Object names whose validated version set exceeded the candidate cap.
/// The scanner retries these names without a version and with destructive
/// healing disabled; this is an explicit bounded continuation, not a
/// version claim.
pub truncated_objects: Vec<String>,
/// Validated candidates beyond the main cap, retained with exact version
/// identities so callers never fall back to a latest-version request.
pub truncated_candidates: Vec<MetaCacheHealCandidate>,
}
impl MetaCacheEntry {
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
let mut wr = Vec::new();
@@ -420,185 +370,6 @@ impl MetaCacheEntries {
})
}
/// Discover validated object/delete-marker versions and safe unversioned
/// inspection candidates in the raw entries without applying read quorum.
/// This is intentionally separate from [`Self::resolve`]: a sub-quorum
/// version is a valid heal target even though it must not participate in
/// normal reads or writes.
///
/// The validated list is bounded and deduplicated by object, version id,
/// and metadata kind; each candidate retains the number of raw disk
/// entries that carried it so callers can classify sub-quorum versions.
/// Entries whose xl.meta cannot be decoded are counted separately for
/// discovery accounting; they never become versionless destructive heal
/// requests and do not consume the validated quota. An
/// [`MetaCacheHealCandidateKind::UnversionedObject`] is always consumed by
/// a non-destructive scanner request.
pub fn discover_heal_candidates(&self, bucket: &str, max_candidates: usize) -> MetaCacheHealDiscovery {
let limit = max_candidates.min(MAX_META_CACHE_HEAL_CANDIDATES);
if limit == 0 || bucket.is_empty() {
return MetaCacheHealDiscovery::default();
}
let mut discovery = MetaCacheHealDiscovery {
candidates: Vec::<MetaCacheHealCandidate>::with_capacity(limit.min(self.0.len())),
unverified_count: 0,
truncated: false,
truncated_objects: Vec::with_capacity(MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS.min(limit)),
truncated_candidates: Vec::new(),
};
let mut seen: HashMap<(String, Option<Uuid>, MetaCacheHealCandidateKind), usize> =
HashMap::with_capacity(limit.min(self.0.len()));
for entry in self.0.iter().flatten() {
if !valid_heal_candidate_name(bucket, entry) {
continue;
}
let meta = match FileMeta::load(&entry.metadata) {
Ok(meta) => meta,
Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
let mut entry_seen = HashSet::new();
for shallow in meta.versions {
let version = match shallow.parse_version_meta() {
Ok(version) if version.valid() => version,
Ok(_) | Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if version.free_version() {
continue;
}
let payload_header = version.header();
if normalize_version_id(shallow.header.version_id) != normalize_version_id(payload_header.version_id)
|| shallow.header.version_type != payload_header.version_type
{
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
let (kind, version_id) = match version.version_type {
VersionType::Object
if version.object.is_some() && version.delete_marker.is_none() && version.legacy_object.is_none() =>
{
match version.object.as_ref().and_then(|object| object.version_id) {
Some(id) if !id.is_nil() => (MetaCacheHealCandidateKind::Object, Some(id)),
Some(_) | None => (MetaCacheHealCandidateKind::UnversionedObject, None),
}
}
VersionType::Delete
if version.delete_marker.is_some() && version.object.is_none() && version.legacy_object.is_none() =>
{
let Some(id) = version.delete_marker.as_ref().and_then(|marker| marker.version_id) else {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
};
if id.is_nil() {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
(MetaCacheHealCandidateKind::DeleteMarker, Some(id))
}
VersionType::Legacy
if version.legacy_object.is_some() && version.object.is_none() && version.delete_marker.is_none() =>
{
let Some(legacy) = version.legacy_object.as_ref() else {
continue;
};
if legacy.version_id.is_empty() {
(MetaCacheHealCandidateKind::UnversionedObject, None)
} else {
let Ok(id) = Uuid::parse_str(&legacy.version_id) else {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
};
if id.is_nil() {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
(MetaCacheHealCandidateKind::Object, Some(id))
}
}
_ => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if normalize_version_id(payload_header.version_id) != version_id {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
// `all_parts=true` is the trust-boundary check for versioned
// candidates. A null/legacy object may still need the old
// non-destructive inspection fallback when its part arrays
// are parseable but incomplete; never use that fallback for
// a candidate carrying a real version id.
let file_info = match version.clone().into_fileinfo(bucket, &entry.name, true) {
Ok(file_info) => file_info,
Err(_) if version_id.is_none() && matches!(kind, MetaCacheHealCandidateKind::UnversionedObject) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
match version.into_fileinfo(bucket, &entry.name, false) {
Ok(file_info) => file_info,
Err(_) => continue,
}
}
Err(_) => {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
};
if file_info.volume != bucket || file_info.name != entry.name {
discovery.unverified_count = discovery.unverified_count.saturating_add(1);
continue;
}
let candidate = MetaCacheHealCandidate {
object: entry.name.clone(),
version_id,
kind,
replica_count: 1,
};
let key = (candidate.object.clone(), candidate.version_id, candidate.kind.clone());
if entry_seen.contains(&key) {
continue;
}
if let Some(index) = seen.get(&key).copied() {
entry_seen.insert(key);
discovery.candidates[index].replica_count = discovery.candidates[index].replica_count.saturating_add(1);
} else if discovery.candidates.len() >= limit {
// Keep the validated candidate list bounded, but retain a
// bounded object-level continuation so the scanner cannot
// silently lose every version of a busy object.
discovery.truncated = true;
if discovery.truncated_objects.len() < MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS
&& !discovery.truncated_objects.iter().any(|object| object == &candidate.object)
{
discovery.truncated_objects.push(candidate.object.clone());
}
if discovery.truncated_objects.iter().any(|object| object == &candidate.object) {
discovery.truncated_candidates.push(candidate);
}
continue;
} else {
entry_seen.insert(key.clone());
seen.insert(key, discovery.candidates.len());
discovery.candidates.push(candidate);
}
}
}
discovery
}
fn resolve_inner(&self, mut params: MetadataResolutionParams, enforce_write_quorum: bool) -> Option<MetaCacheEntry> {
if self.0.is_empty() {
debug!(
@@ -775,33 +546,6 @@ impl MetaCacheEntries {
}
}
fn valid_heal_candidate_name(bucket: &str, entry: &MetaCacheEntry) -> bool {
if bucket.is_empty()
|| entry.name.is_empty()
|| entry.is_dir()
|| (cfg!(windows) && entry.name.contains('\\'))
|| entry.name.chars().any(char::is_control)
{
return false;
}
// Validate raw key components without normalizing them. The scanner maps
// accepted keys to filesystem paths later, so dot components and empty
// internal components must be rejected before that boundary. A final
// empty component is retained for valid keys ending in '/'.
let mut components = entry.name.split('/').peekable();
while let Some(component) = components.next() {
if component == "." || component == ".." || (component.is_empty() && components.peek().is_some()) {
return false;
}
}
true
}
fn normalize_version_id(version_id: Option<Uuid>) -> Option<Uuid> {
version_id.filter(|id| !id.is_nil())
}
#[derive(Debug, Default)]
pub struct MetaCacheEntriesSortedResult {
pub entries: Option<MetaCacheEntriesSorted>,
@@ -1247,7 +991,7 @@ impl<T: Clone + Debug + Send + Sync + 'static> Cache<T> {
mod tests {
use super::*;
use crate::test_data::create_real_xlmeta;
use crate::{FileMetaVersion, MetaDeleteMarker, MetaObjectV1, MetaObjectV1Erasure, MetaObjectV1Stat, TRANSITION_COMPLETE};
use crate::{FileMetaVersion, MetaDeleteMarker, TRANSITION_COMPLETE};
use std::collections::HashMap;
use std::io::Cursor;
use std::sync::{
@@ -1848,381 +1592,6 @@ mod tests {
);
}
#[test]
fn discover_heal_candidates_keeps_sub_quorum_versions_and_deduplicates() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let entries = MetaCacheEntries(vec![
Some(metacache_entry_single_version(1, now, "one")),
Some(metacache_entry_single_version(2, now, "two")),
Some(metacache_entry_single_version(2, now, "two")),
Some(metacache_entry_single_version(3, now, "three")),
]);
let discovery = entries.discover_heal_candidates("bucket", 16);
let ids: std::collections::HashSet<Uuid> = discovery
.candidates
.iter()
.filter_map(|candidate| candidate.version_id)
.collect();
assert_eq!(
ids,
[Uuid::from_u128(1), Uuid::from_u128(2), Uuid::from_u128(3)]
.into_iter()
.collect()
);
assert_eq!(discovery.candidates.len(), 3, "duplicate tied versions must be emitted once");
assert_eq!(
discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(Uuid::from_u128(2)))
.expect("duplicate version should be discovered")
.replica_count,
2
);
}
#[test]
fn discover_heal_candidates_does_not_count_duplicate_versions_within_one_entry() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let mut meta = FileMeta::load(&metacache_entry_single_version(1, now, "duplicate").metadata)
.expect("duplicate fixture should decode");
meta.versions.push(meta.versions[0].clone());
let entry = MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("duplicate metadata should marshal"),
cached: Some(meta),
reusable: false,
};
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 16);
let candidate = discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(Uuid::from_u128(1)))
.expect("duplicate fixture should be discovered");
assert_eq!(candidate.replica_count, 1);
}
#[test]
fn discover_heal_candidates_covers_divergent_quorum_boundaries_n2_n4_n6() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
for (disk_count, quorum) in [(2usize, 1usize), (4, 2), (6, 3)] {
let target_id = Uuid::from_u128(0x1000 + disk_count as u128);
for target_replicas in [quorum.saturating_sub(1), quorum, quorum + 1] {
let entries = (0..disk_count)
.map(|disk| {
let version_id = if disk < target_replicas {
target_id
} else {
Uuid::from_u128(0x2000 + disk as u128)
};
Some(metacache_entry_single_version(version_id.as_u128(), now, "divergent"))
})
.collect();
let discovery = MetaCacheEntries(entries).discover_heal_candidates("bucket", 32);
let target = discovery
.candidates
.iter()
.find(|candidate| candidate.version_id == Some(target_id));
assert_eq!(target.is_some(), target_replicas > 0, "N={disk_count}, replicas={target_replicas}");
if let Some(target) = target {
assert_eq!(target.replica_count, target_replicas);
}
}
}
}
#[test]
fn discover_heal_candidates_separates_delete_markers_and_preserves_unversioned_objects() {
let mut marker_meta = FileMeta::new();
marker_meta
.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(Uuid::from_u128(99)),
deleted: true,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
..Default::default()
})
.expect("delete marker should be added");
let marker = MetaCacheEntry {
name: "object".to_string(),
metadata: marker_meta.marshal_msg().expect("delete marker metadata should marshal"),
cached: Some(marker_meta),
reusable: false,
};
let unversioned_entry = metacache_entry_with_mod_time(
OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"),
"unversioned",
);
let discovery = MetaCacheEntries(vec![Some(marker), Some(unversioned_entry)]).discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::DeleteMarker && candidate.version_id == Some(Uuid::from_u128(99))
}));
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
}
#[test]
fn discover_heal_candidates_rejects_delete_markers_without_ids() {
let mut marker_meta = FileMeta::new();
marker_meta
.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
deleted: true,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
..Default::default()
})
.expect("nil delete marker should be added");
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: marker_meta.marshal_msg().expect("nil marker metadata should marshal"),
cached: Some(marker_meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.kind == MetaCacheHealCandidateKind::DeleteMarker)
);
assert!(discovery.unverified_count >= 1);
}
#[test]
fn discover_heal_candidates_skips_free_versions() {
let object_id = Uuid::from_u128(100);
let free_id = Uuid::from_u128(101);
let mut meta = FileMeta::new();
meta.add_version(FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(object_id),
transition_status: TRANSITION_COMPLETE.to_string(),
transitioned_objname: "remote/object".to_string(),
transition_version_id: Some(Uuid::from_u128(102)),
transition_tier: "WARM".to_string(),
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
})
.expect("transitioned object should be added");
let mut delete = FileInfo {
volume: "bucket".to_string(),
name: "object".to_string(),
version_id: Some(object_id),
mod_time: Some(OffsetDateTime::now_utc()),
..Default::default()
};
delete.set_tier_free_version_id(&free_id.to_string());
meta.delete_version(&delete).expect("free version should be persisted");
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("free version metadata should marshal"),
cached: Some(meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.is_empty());
}
#[test]
fn discover_heal_candidates_preserves_unversioned_legacy_object() {
let legacy = MetaObjectV1 {
version: "1.0.1".to_string(),
format: "xl".to_string(),
stat: MetaObjectV1Stat {
size: 1,
mod_time: Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")),
name: "object".to_string(),
..Default::default()
},
erasure: MetaObjectV1Erasure {
data_blocks: 4,
parity_blocks: 2,
index: 1,
distribution: vec![1, 2, 3, 4, 5, 6],
..Default::default()
},
..Default::default()
};
let version = FileMetaVersion {
version_type: VersionType::Legacy,
legacy_object: Some(legacy),
..Default::default()
};
let mut meta = FileMeta::new();
meta.versions
.push(FileMetaShallowVersion::try_from(version).expect("legacy metadata should marshal"));
let discovery = MetaCacheEntries(vec![Some(MetaCacheEntry {
name: "object".to_string(),
metadata: meta.marshal_msg().expect("legacy metadata should marshal"),
cached: Some(meta),
reusable: false,
})])
.discover_heal_candidates("bucket", 16);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
}
#[test]
fn discover_heal_candidates_rejects_nil_and_malformed_metadata_and_is_bounded() {
let now = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
let mut nil = metacache_entry_single_version(1, now, "nil");
let mut nil_meta = FileMeta::load(&nil.metadata).expect("nil fixture should decode");
let mut nil_version = nil_meta.versions[0]
.parse_version_meta()
.expect("nil fixture version should decode");
nil_version.object.as_mut().expect("object fixture").version_id = Some(Uuid::nil());
nil_meta.versions[0] = FileMetaShallowVersion::try_from(nil_version).expect("nil fixture should marshal");
nil.metadata = nil_meta.marshal_msg().expect("nil fixture metadata should marshal");
let mut mismatched = metacache_entry_single_version(2, now, "mismatched");
let mut mismatched_meta = FileMeta::load(&mismatched.metadata).expect("mismatched fixture should decode");
mismatched_meta.versions[0].header.version_id = Some(Uuid::from_u128(200));
mismatched.metadata = mismatched_meta.marshal_msg().expect("mismatched metadata should marshal");
let mut short_parts = metacache_entry_single_version(3, now, "short-parts");
let mut short_parts_meta = FileMeta::load(&short_parts.metadata).expect("short-parts fixture should decode");
let mut short_parts_version = short_parts_meta.versions[0]
.parse_version_meta()
.expect("short-parts fixture version should decode");
let object = short_parts_version.object.as_mut().expect("object fixture");
object.part_numbers = vec![1];
object.part_actual_sizes = vec![1];
object.part_sizes.clear();
short_parts_meta.versions[0] = FileMetaShallowVersion::try_from(short_parts_version).expect("short-parts should marshal");
short_parts.metadata = short_parts_meta.marshal_msg().expect("short-parts metadata should marshal");
let mut short_unversioned = metacache_entry_with_mod_time(now, "short-unversioned");
let mut short_unversioned_meta =
FileMeta::load(&short_unversioned.metadata).expect("short-unversioned fixture should decode");
let mut short_unversioned_version = short_unversioned_meta.versions[0]
.parse_version_meta()
.expect("short-unversioned version should decode");
let unversioned_object = short_unversioned_version.object.as_mut().expect("unversioned object fixture");
unversioned_object.part_numbers = vec![1];
unversioned_object.part_actual_sizes = vec![1];
unversioned_object.part_sizes.clear();
short_unversioned_meta.versions[0] =
FileMetaShallowVersion::try_from(short_unversioned_version).expect("short-unversioned should marshal");
short_unversioned.metadata = short_unversioned_meta
.marshal_msg()
.expect("short-unversioned metadata should marshal");
let mut malformed = nil.clone();
malformed.name = "malformed".to_string();
malformed.metadata = vec![1, 2, 3];
let entries = MetaCacheEntries(
std::iter::once(Some(nil))
.chain(std::iter::once(Some(mismatched)))
.chain(std::iter::once(Some(short_parts)))
.chain(std::iter::once(Some(short_unversioned)))
.chain(std::iter::once(Some(malformed)))
.chain((0..32).map(|id| Some(metacache_entry_single_version(id + 10, now, "bounded"))))
.collect(),
);
let discovery = entries.discover_heal_candidates("bucket", 5);
assert!(discovery.candidates.len() <= 5);
assert!(discovery.truncated, "bounded discovery must expose dropped candidates");
assert!(
discovery
.truncated_candidates
.iter()
.all(|candidate| candidate.version_id.is_some()),
"overflow candidates must retain exact version identities"
);
assert!(
discovery.truncated_objects.iter().any(|object| object == "object"),
"bounded discovery must expose an object-level safe continuation"
);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::nil()))
);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::from_u128(2)))
);
assert!(
!discovery
.candidates
.iter()
.any(|candidate| candidate.version_id == Some(Uuid::from_u128(3)))
);
assert!(discovery.candidates.iter().any(|candidate| {
candidate.kind == MetaCacheHealCandidateKind::UnversionedObject && candidate.version_id.is_none()
}));
assert!(
discovery.unverified_count >= 1,
"malformed and rejected metadata must remain observable during discovery"
);
for invalid_name in [
"../object",
"./object",
"object/../other",
"object//name",
"object\u{0001}name",
"object\0name",
] {
let mut entry = metacache_entry_single_version(400, now, invalid_name);
entry.name = invalid_name.to_string();
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 5);
assert!(
discovery.candidates.is_empty(),
"invalid key should not become a heal candidate: {invalid_name:?}"
);
}
#[cfg(windows)]
{
let mut entry = metacache_entry_single_version(400, now, "object\\name");
entry.name = "object\\name".to_string();
assert!(
MetaCacheEntries(vec![Some(entry)])
.discover_heal_candidates("bucket", 5)
.candidates
.is_empty(),
"backslash is a path separator on Windows"
);
}
#[cfg(not(windows))]
{
let mut entry = metacache_entry_single_version(400, now, "object\\name");
entry.name = "object\\name".to_string();
assert_eq!(
MetaCacheEntries(vec![Some(entry)])
.discover_heal_candidates("bucket", 5)
.candidates
.len(),
1,
"backslash is object-key data on Unix"
);
}
for valid_name in ["trailing/", "prefix/object"] {
let mut entry = metacache_entry_single_version(401, now, valid_name);
entry.name = valid_name.to_string();
let discovery = MetaCacheEntries(vec![Some(entry)]).discover_heal_candidates("bucket", 5);
assert_eq!(discovery.candidates.len(), 1, "raw S3 key should remain opaque: {valid_name:?}");
}
}
#[test]
fn resolve_rejects_partial_latest_and_returns_committed_previous_metadata() {
let old_mod_time = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
@@ -231,10 +231,6 @@ impl HealTask {
"Heal erasure set format repair skipped because no format heal was required"
);
} else {
let error = e;
if error.is_recoverable_heal() {
return Err(error);
}
error!(
target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_RESULT,
@@ -243,7 +239,7 @@ impl HealTask {
task_id = %self.id,
set_disk_id,
result = "format_failed",
error = %error,
error = %e,
"Heal erasure set failed"
);
{
@@ -251,7 +247,7 @@ impl HealTask {
progress.update_progress(4, 4, 0, 0);
}
return Err(Error::TaskExecutionFailed {
message: format!("Failed to heal disk format for {set_disk_id}: {error}"),
message: format!("Failed to heal disk format for {set_disk_id}: {e}"),
});
}
} else {
@@ -288,9 +284,6 @@ impl HealTask {
Err(Error::TaskCancelled) => return Err(Error::TaskCancelled),
Err(Error::TaskTimeout) => return Err(Error::TaskTimeout),
Err(e) => {
if e.is_recoverable_heal() {
return Err(e);
}
error!(
target: "rustfs::heal::task",
event = EVENT_HEAL_ERASURE_SET_RESULT,
-28
View File
@@ -547,7 +547,6 @@ struct MockStorage {
heal_object_outcome: Mutex<Option<MockHealObjectOutcome>>,
heal_object_outcomes: Mutex<HashMap<String, VecDeque<MockHealObjectOutcome>>>,
format_no_heal_required: Mutex<bool>,
format_error: Mutex<Option<Error>>,
global_format_calls: Mutex<u32>,
replacement_format_calls: Mutex<Vec<(usize, usize, Vec<String>)>>,
replacement_targets_ready: Mutex<bool>,
@@ -868,9 +867,6 @@ impl HealStorageAPI for MockStorage {
async fn heal_format(&self, _dry_run: bool) -> Result<(HealResultItem, Option<Error>)> {
*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();
if no_heal_required {
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::NoHealRequired))))
@@ -2056,30 +2052,6 @@ 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]
async fn erasure_set_bucket_prepass_failure_stops_before_object_heal() {
let temp = TempDir::new().expect("temporary directory should be created");
+67 -116
View File
@@ -43,10 +43,7 @@ use rustfs_common::metrics::{
UpdateCurrentPathFn, current_path_updater, global_metrics,
};
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
use rustfs_filemeta::{
MAX_META_CACHE_HEAL_CANDIDATES, MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS, MetaCacheEntries, MetaCacheEntry,
MetaCacheHealCandidateKind,
};
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration};
use time::OffsetDateTime;
@@ -99,11 +96,6 @@ const METRIC_SCANNER_EXCESS_OBJECT_VERSION_SIZE_TOTAL: &str = "rustfs_scanner_ex
const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders_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_HEAL_DISCOVERY_CANDIDATES_TOTAL: &str = "rustfs_scanner_heal_discovery_candidates_total";
const METRIC_SCANNER_HEAL_DISCOVERY_SUB_QUORUM_TOTAL: &str = "rustfs_scanner_heal_discovery_sub_quorum_total";
const METRIC_SCANNER_HEAL_DISCOVERY_UNVERIFIED_TOTAL: &str = "rustfs_scanner_heal_discovery_unverified_total";
const METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL: &str = "rustfs_scanner_heal_discovery_queued_total";
const METRIC_SCANNER_HEAL_DISCOVERY_TRUNCATED_TOTAL: &str = "rustfs_scanner_heal_discovery_truncated_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
@@ -891,7 +883,7 @@ impl FolderScanner {
object: Option<String>,
version_id: Option<String>,
request: HealChannelRequest,
) -> Result<HealAdmissionResult, ScannerError> {
) -> Result<(), ScannerError> {
let candidate_type = pending_scanner_heal_candidate_type(kind);
let priority = request.priority;
let scan_mode = request.scan_mode.unwrap_or(self.scan_mode);
@@ -919,7 +911,7 @@ impl FolderScanner {
error = %err,
"Scanner deferred heal request after channel error"
);
return Ok(HealAdmissionResult::Full);
return Ok(());
}
};
self.update_pending_scanner_heal_after_admission(
@@ -931,7 +923,7 @@ impl FolderScanner {
result,
);
if result.is_admitted() {
return Ok(result);
return Ok(());
}
record_high_priority_heal_escalation(candidate_type, priority, result);
@@ -952,7 +944,7 @@ impl FolderScanner {
state = "high_priority_not_admitted",
"Scanner high-priority heal admission failed"
);
Ok(result)
Ok(())
}
pub fn set_heal_object_select(&mut self, prob: u32) {
@@ -1744,7 +1736,14 @@ impl FolderScanner {
break;
}
let mut previous_bucket = String::new();
let mut resolver = MetadataResolutionParams {
dir_quorum: self.disks_quorum,
obj_quorum: self.disks_quorum,
bucket: "".to_string(),
strict: false,
..Default::default()
};
for name in abandoned_children {
if !self.should_heal().await {
break;
@@ -1752,7 +1751,7 @@ impl FolderScanner {
let (bucket, prefix) = path2_bucket_object(name.as_str());
if bucket != previous_bucket {
if bucket != resolver.bucket {
self.send_required_scanner_heal_request(
PendingScannerHealKind::Bucket,
bucket.clone(),
@@ -1761,9 +1760,10 @@ impl FolderScanner {
build_bucket_heal_request(bucket.clone(), HealChannelPriority::High),
)
.await?;
previous_bucket = bucket.clone();
}
resolver.bucket = bucket.clone();
let child_ctx = ctx.child_token();
let (agreed_tx, mut agreed_rx) = mpsc::channel::<String>(1);
@@ -1880,8 +1880,6 @@ impl FolderScanner {
let mut agreed_closed = false;
let mut partial_closed = false;
let mut finished_closed = false;
let mut seen_heal_candidates: HashSet<(String, Option<String>, MetaCacheHealCandidateKind)> = HashSet::new();
let mut seen_truncated_objects: HashSet<String> = HashSet::new();
loop {
if agreed_closed && partial_closed && finished_closed {
@@ -1906,112 +1904,65 @@ impl FolderScanner {
break;
}
let discovery = entries.discover_heal_candidates(&bucket, MAX_META_CACHE_HEAL_CANDIDATES);
counter!(METRIC_SCANNER_HEAL_DISCOVERY_CANDIDATES_TOTAL)
.increment(u64::try_from(discovery.candidates.len()).unwrap_or(u64::MAX));
counter!(METRIC_SCANNER_HEAL_DISCOVERY_SUB_QUORUM_TOTAL).increment(
u64::try_from(
discovery
.candidates
.iter()
.filter(|candidate| candidate.replica_count < disks_quorum)
.count(),
)
.unwrap_or(u64::MAX),
);
counter!(METRIC_SCANNER_HEAL_DISCOVERY_UNVERIFIED_TOTAL).increment(
u64::try_from(discovery.unverified_count).unwrap_or(u64::MAX),
);
if discovery.truncated {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_TRUNCATED_TOTAL).increment(1);
let Some(entry) = resolve_object_heal_entry(&entries, resolver.clone()) else {
continue;
};
(self.update_current_path)(&entry.name).await;
if entry.is_dir() {
continue;
}
for candidate in discovery.candidates {
let sub_quorum_candidate = candidate.replica_count < disks_quorum;
let version_id = candidate.validated_version().map(|id| id.to_string());
let identity = (candidate.object.clone(), version_id.clone(), candidate.kind.clone());
if seen_heal_candidates.len() >= MAX_META_CACHE_HEAL_CANDIDATES
&& !seen_heal_candidates.contains(&identity)
{
continue;
}
if !seen_heal_candidates.insert(identity) {
continue;
}
let request = if candidate.is_unversioned() {
build_non_destructive_object_heal_request(
bucket.clone(),
candidate.object.clone(),
self.scan_mode,
HealChannelPriority::High,
)
} else {
build_object_heal_request(
bucket.clone(),
candidate.object.clone(),
version_id.clone(),
self.scan_mode,
HealChannelPriority::High,
)
};
(self.update_current_path)(&candidate.object).await;
let admission = self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(candidate.object.clone()),
version_id.clone(),
request,
)
.await?;
if admission.is_admitted() {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL).increment(1);
} else if sub_quorum_candidate {
self.mark_pending_scanner_heal_reason(
PendingScannerHealKind::Object,
&bucket,
Some(&candidate.object),
version_id.as_deref(),
"sub_quorum_metadata",
let fivs = match entry.file_info_versions(&bucket) {
Ok(fivs) => fivs,
Err(e) => {
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %bucket,
entry = %entry.name,
state = "file_info_versions_failed",
error = %e,
"Scanner list_path_raw failed to resolve file versions"
);
}
found_objects = true;
}
// Candidates beyond the main cap remain exact
// version requests; never downgrade them to a
// latest-version (version_id=None) heal.
for candidate in discovery.truncated_candidates {
let version_id = candidate.validated_version().map(|id| id.to_string());
let identity = (candidate.object.clone(), version_id.clone(), candidate.kind.clone());
if seen_truncated_objects.len() >= MAX_META_CACHE_HEAL_TRUNCATED_OBJECTS
&& !seen_truncated_objects.contains(&candidate.object)
{
continue;
}
seen_truncated_objects.insert(candidate.object.clone());
if !seen_heal_candidates.insert(identity) {
continue;
}
let request = build_object_heal_request(
bucket.clone(),
candidate.object.clone(),
version_id.clone(),
self.scan_mode,
HealChannelPriority::High,
);
(self.update_current_path)(&candidate.object).await;
let admission = self
.send_required_scanner_heal_request(
self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(candidate.object.clone()),
version_id,
request,
Some(entry.name.clone()),
None,
build_object_heal_request(
bucket.clone(),
entry.name.clone(),
None,
self.scan_mode,
HealChannelPriority::High,
),
)
.await?;
if admission.is_admitted() {
counter!(METRIC_SCANNER_HEAL_DISCOVERY_QUEUED_TOTAL).increment(1);
found_objects = true;
continue;
}
};
for fiv in fivs.versions {
let version_id = fiv.version_id.and_then(|v| if v.is_nil() { None } else { Some(v.to_string()) });
self.send_required_scanner_heal_request(
PendingScannerHealKind::Object,
bucket.clone(),
Some(entry.name.clone()),
version_id.clone(),
build_object_heal_request(
bucket.clone(),
entry.name.clone(),
version_id,
self.scan_mode,
HealChannelPriority::High,
),
)
.await?;
found_objects = true;
}
@@ -13,8 +13,6 @@
// limitations under the License.
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
use super::*;
#[cfg(test)]
use rustfs_filemeta::MetadataResolutionParams;
/// Cached folder information for scanning
#[derive(Clone, Debug)]
@@ -90,22 +88,6 @@ pub(super) fn build_object_heal_request(
}
}
/// Build the versionless inspection request used when discovery cannot prove
/// a destructive version identity (for example an unversioned object or a
/// bounded candidate overflow). The explicit flag is the fail-closed safety
/// boundary; callers must not reconstruct it with the destructive default.
pub(super) fn build_non_destructive_object_heal_request(
bucket: String,
object: String,
scan_mode: HealScanMode,
priority: HealChannelPriority,
) -> HealChannelRequest {
let mut request = build_object_heal_request(bucket, object, None, scan_mode, priority);
request.remove_corrupted = Some(false);
request
}
#[cfg(test)]
pub(super) fn resolve_object_heal_entry(
entries: &MetaCacheEntries,
resolver: MetadataResolutionParams,
+7 -39
View File
@@ -105,29 +105,6 @@ impl FolderScanner {
}
}
/// Preserve the discovery reason when a candidate could not be admitted
/// immediately. The existing string field is intentionally reused so the
/// scanner's map-encoded cache schema stays backward compatible.
pub(super) fn mark_pending_scanner_heal_reason(
&mut self,
kind: PendingScannerHealKind,
bucket: &str,
object: Option<&str>,
version_id: Option<&str>,
reason: &str,
) {
if let Some(entry) = self
.new_cache
.info
.pending_heals
.iter_mut()
.find(|entry| pending_scanner_heal_matches(entry, kind, bucket, object, version_id))
{
entry.last_admission_reason = reason.to_string();
self.sync_pending_heals();
}
}
pub(super) fn prune_pending_scanner_heals(&mut self) {
let now = Self::now_secs();
let before_expiry = self.new_cache.info.pending_heals.len();
@@ -328,22 +305,13 @@ pub(super) fn build_pending_scanner_heal_request(entry: &PendingScannerHeal) ->
match entry.kind {
PendingScannerHealKind::Bucket => Some(build_bucket_heal_request(entry.bucket.clone(), HealChannelPriority::High)),
PendingScannerHealKind::Object => entry.object.as_ref().map(|object| {
if entry.version_id.is_none() {
build_non_destructive_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
} else {
build_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.version_id.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
}
build_object_heal_request(
entry.bucket.clone(),
object.clone(),
entry.version_id.clone(),
entry.scan_mode,
HealChannelPriority::High,
)
}),
}
}
+6 -106
View File
@@ -17,7 +17,7 @@ use crate::SCANNER_SLEEPER;
use super::*;
use crate::storage_api::VersionPurgeStatusType;
use crate::{DiskOption, Endpoint, STORAGE_FORMAT_FILE, TierStats, new_disk, storageclass};
use rustfs_filemeta::{FileInfo, FileMeta, MetadataResolutionParams};
use rustfs_filemeta::{FileInfo, FileMeta};
use std::io::Write;
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
@@ -982,21 +982,6 @@ fn test_build_object_heal_request_omits_nil_version_id() {
assert_eq!(request.recreate_missing, Some(false));
}
#[test]
fn test_build_non_destructive_object_heal_request_disables_removal() {
let request = build_non_destructive_object_heal_request(
"bucket".to_string(),
"path/to/object".to_string(),
HealScanMode::Deep,
HealChannelPriority::High,
);
assert_eq!(request.object_version_id, None);
assert_eq!(request.remove_corrupted, Some(false));
assert_eq!(request.recreate_missing, Some(false));
assert_eq!(request.source, HealRequestSource::Scanner);
}
#[test]
fn test_build_bucket_heal_request_disables_recreate_for_scanner() {
let request = build_bucket_heal_request("bucket".to_string(), HealChannelPriority::Low);
@@ -1136,42 +1121,6 @@ fn test_pending_heal_reconstructs_object_request_with_version() {
assert_eq!(request.source, HealRequestSource::Scanner);
}
#[test]
fn test_pending_heal_reconstructs_unversioned_request_without_removal() {
let pending = pending_heal(PendingScannerHealKind::Object, "bucket", Some("object"), None, 1, 1);
let request = build_pending_scanner_heal_request(&pending).expect("unversioned object request should rebuild");
assert!(request.object_version_id.is_none());
assert_eq!(request.remove_corrupted, Some(false));
assert_eq!(request.recreate_missing, Some(false));
}
#[tokio::test]
async fn test_pending_heal_reason_preserves_sub_quorum_discovery() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
scanner.update_pending_scanner_heal_after_admission(
PendingScannerHealKind::Object,
"bucket",
Some("object"),
Some("version-a"),
HealScanMode::Deep,
HealAdmissionResult::Full,
);
scanner.mark_pending_scanner_heal_reason(
PendingScannerHealKind::Object,
"bucket",
Some("object"),
Some("version-a"),
"sub_quorum_metadata",
);
assert_eq!(scanner.new_cache.info.pending_heals.len(), 1);
assert_eq!(scanner.new_cache.info.pending_heals[0].last_admission_reason, "sub_quorum_metadata");
}
#[test]
fn test_pending_heal_retry_candidates_respect_cap_and_order() {
let pending: Vec<PendingScannerHeal> = (0..(MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET + 2))
@@ -1374,20 +1323,6 @@ fn metadata_for_object(bucket: &str, object: &str) -> Vec<u8> {
meta.marshal_msg().expect("test metadata should marshal")
}
fn metadata_for_object_version(bucket: &str, object: &str, version_id: Option<Uuid>) -> Vec<u8> {
let mut file_info = FileInfo::new(object, 4, 2);
file_info.volume = bucket.to_string();
file_info.name = object.to_string();
file_info.version_id = version_id;
file_info.versioned = version_id.is_some();
file_info.mod_time = Some(OffsetDateTime::now_utc());
file_info.size = 1;
let mut meta = FileMeta::new();
meta.add_version(file_info).expect("test metadata version should be accepted");
meta.marshal_msg().expect("test metadata should marshal")
}
async fn write_test_object_metadata(root: &std::path::Path, bucket: &str, object: &str) {
write_test_object_metadata_bytes(root, bucket, object, &metadata_for_object(bucket, object)).await;
}
@@ -1791,21 +1726,12 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
let heal_starts = Arc::new(AtomicUsize::new(0));
let heal_starts_clone = heal_starts.clone();
let healed_versions = Arc::new(Mutex::new(Vec::<Option<String>>::new()));
let healed_versions_clone = healed_versions.clone();
let mut heal_rx =
rustfs_common::heal_channel::init_heal_channel().expect("heal channel should initialize once for scanner tests");
let _heal_responder = tokio::spawn(async move {
while let Some(command) = heal_rx.recv().await {
if let rustfs_common::heal_channel::HealChannelCommand::Start {
request, response_tx, ..
} = command
{
if let rustfs_common::heal_channel::HealChannelCommand::Start { response_tx, .. } = command {
heal_starts_clone.fetch_add(1, Ordering::Relaxed);
healed_versions_clone
.lock()
.expect("heal version capture lock should not be poisoned")
.push(request.object_version_id);
let _ = response_tx.send(Ok(HealAdmissionResult::Accepted));
}
}
@@ -1813,18 +1739,13 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
let bucket = "src-archive";
let object = "snapshots/37b3f20d941e2f5e6d99114d9bb2f3e67a8a2e5c9c4c5a1b0d6e7f8091a2b3c4";
let orphan_version = Uuid::from_u128(0x1934);
let shared_version = Uuid::from_u128(0x1935);
let orphan_metadata = metadata_for_object_version(bucket, object, Some(orphan_version));
let shared_metadata = metadata_for_object_version(bucket, object, Some(shared_version));
write_test_object_metadata_bytes(&temp_dir, bucket, object, &orphan_metadata).await;
let mut expected_metadata = vec![(temp_dir.join(bucket).join(object).join("xl.meta"), orphan_metadata.clone())];
let metadata = metadata_for_object(bucket, object);
write_test_object_metadata_bytes(&temp_dir, bucket, object, &metadata).await;
let mut disks = vec![scanner.local_disk.clone()];
for disk_name in ["disk2", "disk3", "disk4"] {
let disk_root = temp_dir.join(disk_name);
write_test_object_metadata_bytes(&disk_root, bucket, object, &shared_metadata).await;
expected_metadata.push((disk_root.join(bucket).join(object).join("xl.meta"), shared_metadata.clone()));
write_test_object_metadata_bytes(&disk_root, bucket, object, &metadata).await;
let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("failed to create extra disk endpoint");
let disk = new_disk(
&endpoint,
@@ -1873,29 +1794,8 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() {
.new_cache
.checked_flatten(bucket)
.expect("healed cache must contain canonical child links");
// The fixture intentionally exposes two divergent version histories, so
// the scanner keeps both logical versions visible while discovering heals.
assert_eq!(root.objects, 2);
assert_eq!(root.objects, 1);
assert!(heal_starts.load(Ordering::Relaxed) > 0, "test must execute the heal child-link path");
let orphan_version_text = orphan_version.to_string();
assert!(
healed_versions
.lock()
.expect("heal version capture lock should not be poisoned")
.iter()
.any(|version| version.as_deref() == Some(orphan_version_text.as_str())),
"sub-quorum orphan version must be submitted as an exact heal candidate"
);
for (path, expected) in expected_metadata {
assert_eq!(
tokio::fs::read(&path)
.await
.expect("scanner discovery must not delete metadata"),
expected,
"scanner discovery must not modify candidate metadata: {}",
path.display()
);
}
}
#[tokio::test]
-12
View File
@@ -245,18 +245,6 @@ impl TestECStoreEnvBuilder {
.await
.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 {
let buckets_list = ecstore
.list_bucket(&BucketOptions {
+3
View File
@@ -43,6 +43,9 @@ allow-git = [
# RustFS fork carrying presigned expiry and constant-time authentication fixes.
# owner: rustfs-maintainers review: 2026-10
"https://github.com/rustfs/s3s.git",
# MiMalloc fork pinned for hotpath allocation counting support.
# owner: houseme review: 2026-10
"https://github.com/xonatius/mimalloc_rust.git",
]
[bans]
-149
View File
@@ -1,149 +0,0 @@
# CI gate matrix
This file is the source of truth for which validation runs on each event, its
configured wall-clock budget, and whether it can block a merge. Test taxonomy,
naming, and nextest serialization rules remain in [README.md](README.md); e2e
membership and counts remain in
[e2e-suite-inventory.md](e2e-suite-inventory.md).
The distinction between **required** and **report-only** is load-bearing:
a failing job blocks a merge only when its exact check name is present in the
live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
does not make a job required by itself.
## Required merge checks
The live `main` ruleset (`6436880`) currently requires exactly these contexts:
| Required context | Producer | Validation |
|---|---|---|
| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
For pull requests limited to the paths excluded by the main CI workflow,
`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
`Test and Lint` under the same names. It runs the real quick checks and the
planning-document guard; it does not claim that Rust compilation or runtime
tests ran. Despite the workflow name, these paths also include selected deploy,
workflow, and lock files.
Verify the live rule rather than trusting this snapshot before changing merge
policy:
```bash
gh api repos/rustfs/rustfs/rulesets/6436880 \
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
```
The ruleset currently has `strict_required_status_checks_policy=false`.
`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
them, but `End-to-End Tests (full merge gate)` is not currently a required
context. Therefore the repository is prepared to test a merge-queue SHA, but
the workflow alone does not prove that every merge passed that lane.
## Pull request and merge matrix
Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
means the result is visible and actionable but is not in the live required
context list.
| Event | Validation | Budget | Merge status | Reproduction |
|---|---|---:|---|---|
| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
instead of adding a second ad-hoc selector. Before a profile runs,
`scripts/check_test_wiring.py` compares its exact membership to the committed
digest so a silent test drop fails closed.
## Scheduled and manual validation
Scheduled lanes are independent fault domains. They do not block a pull
request, but their workflow-local gate can fail the run and scheduled failures
are routed to the shared failure-issue action. The scheduled-validation
watchdog and freshness workflow separately detect incomplete runs and missing
schedules.
| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
|---|---|---:|---|---|
| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
runs are debugging evidence and intentionally do not open scheduled-failure
issues. A manual performance run may explicitly allow a known regression; that
override must not be treated as an ordinary passing baseline.
## Release validation
Release validation is post-merge and tag-driven; it does not substitute for a
pull-request gate.
| Event | Validation | Budget | Result |
|---|---|---:|---|
| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
are backfill/debug paths and do not prove the automatic `workflow_run` chain.
## Evidence requirements
A green check is useful only when it proves the intended behavior ran:
- Record the exact commit SHA and run URL.
- Separate product failure from runner prerequisites, service readiness, and
cancellation. Repair the precondition, then rerun the exact workload.
- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
server logs where the workflow provides them.
- For a bug fix or a new fault checker, provide sensitivity evidence: the old
behavior or an intentional mutation must fail the new oracle, and the fixed
behavior must pass it.
- Never promote a report-only lane to required from one green run. Require at
least 14 days and 30 representative pull requests with at least 99% complete
execution, then update the ruleset and this table together.
## Change checklist
Update this file in the same pull request when any of these change:
- workflow triggers, job names, timeouts, or nextest profile ownership;
- required status contexts or strict/merge-queue policy;
- scheduled cadence, alert routing, artifact contract, or local reproduction;
- report-only versus gating semantics.
Do not copy per-module test counts here. Update
[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
digest instead.
+4 -4
View File
@@ -322,7 +322,7 @@ thiserror = { workspace = true }
tracing.workspace = true
url = { workspace = true }
urlencoding = { workspace = true }
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }
zip = { workspace = true }
libc = { workspace = true }
rand = { workspace = true, features = ["serde"] }
@@ -336,16 +336,16 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
rustfs-mimalloc = { workspace = true }
mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
rustfs-mimalloc-sys.workspace = true
libmimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }
serial_test = { workspace = true }
tempfile = { workspace = true }
aws-config = { workspace = true }
+32 -121
View File
@@ -41,7 +41,7 @@ use crate::admin::storage_api::config::save_admin_config;
use crate::admin::storage_api::contract::bucket::{
BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp,
};
use crate::admin::storage_api::error::{Error as StorageError, is_err_bucket_not_found};
use crate::admin::storage_api::error::Error as StorageError;
use crate::admin::storage_api::runtime::ECStore;
use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body};
use crate::auth::constant_time_eq;
@@ -55,7 +55,6 @@ use crate::storage::storage_api::{
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use futures::StreamExt;
use hmac::{Hmac, Mac};
use http::header::{CONTENT_TYPE, HOST};
use http::{HeaderMap, HeaderValue, Uri};
@@ -2097,18 +2096,6 @@ async fn remote_add_preflight_info(site: &PeerSite) -> S3Result<SiteReplicationA
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(
&client,
@@ -2219,30 +2206,20 @@ fn site_replication_bootstrap_token(uri: &Uri) -> Option<String> {
query_pairs(uri).get("bootstrapToken").cloned()
}
/// 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 {
fn bootstrap_bucket_make_op_path(bucket: &SRBucketInfo) -> String {
let mut query = form_urlencoded::Serializer::new(String::new());
query.append_pair("bucket", bucket);
query.append_pair("operation", SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING);
query.append_pair("versioningEnabled", "true");
if let Some(created_at) = created_at {
query.append_pair("createdAt", created_at);
query.append_pair("bucket", &bucket.bucket);
query.append_pair("operation", "make-with-versioning");
if let Some(created_at) = bucket
.created_at
.and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok())
{
query.append_pair("createdAt", &created_at);
}
if lock_enabled {
if bucket.object_lock_config.is_some() {
query.append_pair("lockEnabled", "true");
}
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())
format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?{}", query.finish())
}
fn bootstrap_bucket_meta_item(bucket: &SRBucketInfo, item_type: &str, updated_at: Option<OffsetDateTime>) -> SRBucketMeta {
@@ -4269,7 +4246,16 @@ async fn broadcast_site_replication_make_bucket(
.format(&time::format_description::well_known::Rfc3339)
.unwrap_or_default();
let path = make_with_versioning_bucket_op_path(bucket, Some(&created_at), lock_enabled);
let path = {
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 {
with_site_replication_bootstrap_token(&path, token)
} else {
@@ -10220,25 +10206,13 @@ 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 {}
#[async_trait::async_trait]
impl Operation for SiteReplicationDevNullHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?;
drain_site_replication_devnull(req.input).await?;
let _ = read_plain_admin_body(req.input).await?;
Ok(empty_response(StatusCode::NO_CONTENT))
}
}
@@ -10497,19 +10471,6 @@ 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 {}
#[async_trait::async_trait]
@@ -10609,18 +10570,16 @@ impl Operation for SRPeerBucketOpsHandler {
.map_err(ApiError::from)?;
}
"purge-deleted-bucket" => {
purge_deleted_bucket_result(
store
.delete_bucket(
&bucket,
&DeleteBucketOptions {
force: true,
srdelete_op: SRBucketDeleteOp::Purge,
..Default::default()
},
)
.await,
)?;
let _ = store
.delete_bucket(
&bucket,
&DeleteBucketOptions {
force: true,
srdelete_op: SRBucketDeleteOp::Purge,
..Default::default()
},
)
.await;
}
_ => return Err(s3_error!(InvalidRequest, "unsupported site replication bucket operation")),
}
@@ -13966,54 +13925,6 @@ mod tests {
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]
#[serial]
async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() {
+5 -24
View File
@@ -13,9 +13,9 @@
// limitations under the License.
use rustfs_madmin::{PeerInfo, SyncStatus};
use std::collections::BTreeMap;
use std::collections::{BTreeMap, hash_map::DefaultHasher};
use std::hash::{Hash, Hasher};
use url::Url;
use uuid::Uuid;
fn has_http_scheme(endpoint: &str) -> bool {
endpoint.get(..7).is_some_and(|prefix| prefix.eq_ignore_ascii_case("http://"))
@@ -66,12 +66,10 @@ pub fn site_identity_key(endpoint: &str) -> String {
.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 {
Uuid::new_v5(&Uuid::NAMESPACE_URL, canonical_endpoint(endpoint).as_bytes()).to_string()
let mut hasher = DefaultHasher::new();
endpoint.hash(&mut hasher);
format!("{:016x}", hasher.finish())
}
pub fn same_identity_endpoint(left: &str, right: &str) -> bool {
@@ -176,23 +174,6 @@ 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]
fn canonical_endpoint_accepts_case_insensitive_scheme() {
assert_eq!(
+1 -2
View File
@@ -51,7 +51,7 @@ mod ecstore_disk {
}
mod ecstore_error {
pub(crate) use crate::storage::storage_api::ecstore_error::{StorageError, is_err_bucket_not_found};
pub(crate) use crate::storage::storage_api::ecstore_error::StorageError;
}
#[allow(unused_imports)]
@@ -919,7 +919,6 @@ pub(crate) mod contract {
}
pub(crate) mod error {
pub(crate) use super::ecstore_error::is_err_bucket_not_found;
pub(crate) use super::{Error, StorageError};
}
+7 -1
View File
@@ -369,8 +369,14 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> {
rustfs_mimalloc::MiMalloc::collect(force);
// SAFETY: `mi_collect` is provided by the active global allocator backend
// on this target family. It is explicitly intended to reclaim retained
// pages/segments and does not require additional invariants from the caller.
unsafe {
libmimalloc_sys::mi_collect(force);
}
Ok(())
}
+8 -10
View File
@@ -26,22 +26,22 @@ struct MiMallocAllocator;
unsafe impl GlobalAlloc for MiMallocAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
unsafe { mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
}
}
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator]
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,9 +71,8 @@ mod tests {
allocation.extend_from_slice(&[7_u8; 64]);
assert_eq!(allocation.len(), 64);
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: the live Vec pointer is valid to inspect for heap ownership.
assert!(unsafe { heap.contains(allocation.as_ptr()) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
}
#[test]
@@ -86,13 +85,12 @@ mod tests {
let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
let allocator = super::MiMallocAllocator;
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: The pointer is checked for null before use and later released
// through the same allocator with the corresponding layout.
let ptr = unsafe { allocator.alloc_zeroed(layout) };
assert!(!ptr.is_null());
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -104,7 +102,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { heap.contains(grown_ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+36 -19
View File
@@ -17,7 +17,10 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split,
};
use serde::Serialize;
#[cfg(any(test, not(target_os = "windows")))]
use serde_json::Value;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
@@ -228,18 +231,7 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
let json = rustfs_mimalloc::MiMalloc::stats_json();
if json.is_empty() {
return None;
}
let observation = parse_mimalloc_stats_json(&json)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -250,6 +242,7 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
@@ -261,6 +254,7 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
@@ -277,10 +271,12 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current")
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
metrics
.iter()
@@ -289,6 +285,7 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
.filter(|value| *value > 0)
}
#[cfg(any(test, not(target_os = "windows")))]
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?;
let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -315,6 +312,33 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
let observation = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
libmimalloc_sys::mi_free(stats_ptr.cast());
observation?
};
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -542,13 +566,6 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[test]
fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
let snapshot = super::read_allocator_memory_snapshot();
#[cfg(not(target_os = "windows"))]
assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
}
#[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false);