Compare commits

..

5 Commits

Author SHA1 Message Date
houseme d695c94661 Merge branch 'main' into cxymds/fix-1937-heal-resume-gc 2026-08-23 12:13:36 +08:00
Zhengchao An b6ba89d9e4 docs(testing): document CI gate matrix (#6412) 2026-08-23 12:09:06 +08:00
houseme 648d5166e2 feat(allocator): replace mimalloc/libmimalloc-sys with rustfs-mimalloc/rustfs-mimalloc-sys (#6404)
Replace the upstream xonatius/mimalloc_rust.git fork (mimalloc + libmimalloc-sys)
with the published rustfs-mimalloc (v0.5.0) and rustfs-mimalloc-sys (v0.5.0) crates
from crates.io.

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

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

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

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

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

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-23 12:07:20 +08:00
马登山 357b42406c fix(heal): add bounded resume artifact inspection 2026-08-23 07:24:54 +08:00
18 changed files with 942 additions and 190 deletions
+2 -1
View File
@@ -30,7 +30,8 @@ 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 gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
+2
View File
@@ -70,6 +70,8 @@ make pre-pr
> For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
> For the 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
+22 -27
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.3"
version = "1.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,12 +2522,6 @@ 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"
@@ -5988,15 +5982,6 @@ 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"
@@ -6397,14 +6382,6 @@ 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"
@@ -9162,13 +9139,11 @@ 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",
@@ -9204,6 +9179,8 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9875,6 +9852,24 @@ dependencies = [
"tokio",
]
[[package]]
name = "rustfs-mimalloc"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
dependencies = [
"rustfs-mimalloc-sys",
]
[[package]]
name = "rustfs-mimalloc-sys"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
dependencies = [
"cc",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.3"
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
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"] }
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+1 -6
View File
@@ -278,9 +278,4 @@ listed by `cargo nextest list -p e2e_test`. Regenerate it when adding or
moving e2e tests so acceptance numbers in the test-strategy issues
(backlog#1147#1155) stay auditable. When a profile membership change is
intentional, review its JSON listing before updating the matching
`.config/e2e-*-selection.txt` test-ID digest. Update only the platform that
produced the listing:
```bash
python3 scripts/check_test_wiring.py --update-profile e2e-full /path/to/listing.json linux
```
`.config/e2e-*-selection.txt` test-ID digest.
+1 -14
View File
@@ -2124,26 +2124,13 @@ 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 {
+1 -1
View File
@@ -3194,7 +3194,7 @@ impl ECStore {
// Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let mut accounting = vec![None; objects.len()];
let accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
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(|left, right| right.idx.cmp(&left.idx));
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+49 -1
View File
@@ -14,7 +14,7 @@
use crate::heal::{
progress::{HealProgress, HealStatistics},
resume::{ReplacementPhase, ResumeManager, ResumeState, ResumeUtils},
resume::{ReplacementPhase, ResumeGc, ResumeManager, ResumeState, ResumeUtils},
storage::HealStorageAPI,
task::{HealOptions, HealPriority, HealRequest, HealTask, HealTaskStatus, HealType, demote_to_debug_when},
};
@@ -53,9 +53,11 @@ const EVENT_HEAL_MAINLINE_THROTTLE: &str = "heal_mainline_throttle";
const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state";
const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state";
const EVENT_HEAL_UNCLEAN_SHUTDOWN: &str = "heal_unclean_shutdown";
const EVENT_HEAL_RESUME_GC: &str = "heal_resume_gc";
const LEGACY_ROOT_HEAL_PATH: &str = ".";
const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3;
const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30);
const RESUME_GC_INTERVAL: Duration = Duration::from_secs(60 * 60);
// Admission/scheduler outcomes for per-object requests (Object/Metadata/
// ECDecode) log via demote_to_debug_when! — MRF, autoheal, and scanner
@@ -1150,6 +1152,49 @@ impl HealManager {
Ok(())
}
/// Start the bounded resume-state inspector. Destructive GC remains
/// disabled until the durable owner/CAS contract from backlog#1927 is
/// available; this task therefore cannot remove an active or stale file.
async fn start_resume_gc(&self) {
let cancel = self.cancel_token.clone();
tokio::spawn(async move {
let mut gc_by_disk = HashMap::<String, ResumeGc>::new();
let mut ticker = interval(RESUME_GC_INTERVAL);
loop {
tokio::select! {
_ = cancel.cancelled() => break,
_ = ticker.tick() => {
let disks = {
let local_disk_map = local_disk_map_read().await;
local_disk_map.values().flatten().cloned().collect::<Vec<_>>()
};
for disk in disks {
let disk_key = disk.endpoint().to_string();
let gc = gc_by_disk.entry(disk_key).or_default();
tokio::select! {
_ = cancel.cancelled() => return,
result = gc.inspect_disk(&disk) => {
if let Err(error) = result {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_RESUME_GC,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "inspect_failed",
endpoint = %disk.endpoint(),
error = %error,
"Heal resume GC inspection failed"
);
}
}
}
}
}
}
}
});
}
/// Create new HealManager
pub fn new(storage: Arc<dyn HealStorageAPI>, config: Option<HealConfig>) -> Self {
Self::new_with_workload_provider(storage, config, None)
@@ -1215,6 +1260,9 @@ impl HealManager {
// competing task for the same set.
self.process_unclean_shutdown().await;
// Inspect resume artifacts in a bounded, fail-closed background task.
self.start_resume_gc().await;
// start auto disk scanner to heal unformatted disks
if self.config.read().await.enable_auto_heal {
self.start_auto_disk_scanner().await?;
+2
View File
@@ -28,10 +28,12 @@ use super::{
};
mod checkpoint;
mod gc;
mod replacement;
mod utils;
pub use checkpoint::{CheckpointManager, ResumeCheckpoint};
pub(crate) use gc::ResumeGc;
pub(crate) use replacement::replacement_target_identities_match;
use replacement::replacement_targets_match_identities;
pub use replacement::{
+666
View File
@@ -0,0 +1,666 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Bounded inspection of heal resume artifacts.
//!
//! The durable owner/CAS and quarantine primitives belong to backlog #1927 and
//! are not part of the current base revision. This module is therefore
//! deliberately inspect-only. In particular, it must never turn an age check
//! into a delete: ordinary heal writers still publish raw files on this base,
//! so a GC-side compare-and-delete would not fence a concurrent claim.
use metrics::counter;
use std::{
collections::BTreeMap,
path::{Component, Path},
time::{SystemTime, UNIX_EPOCH},
};
use tokio::io::AsyncReadExt;
use super::super::{BUCKET_META_PREFIX, DiskError, DiskStore, RUSTFS_META_BUCKET, storage_api::owner::EcstoreDiskAPI};
use super::{
LEGACY_REPLACEMENT_RECOVERY_MARKER_FILE, REPLACEMENT_COMPLETION_PROOF_FILE, REPLACEMENT_INTENT_FILE,
REPLACEMENT_INTENT_SEAL_FILE, RESUME_CHECKPOINT_FILE, RESUME_PROGRESS_FILE, RESUME_STATE_FILE, ResumeCheckpoint, ResumeState,
checkpoint::CURRENT_CHECKPOINT_SCHEMA,
};
use crate::{Error, Result};
const DEFAULT_ENTRY_BUDGET: usize = 256;
const DEFAULT_BYTE_BUDGET: usize = 4 * 1024 * 1024;
const GC_METRIC: &str = "rustfs_heal_resume_gc_inspected_total";
const GC_ERROR_METRIC: &str = "rustfs_heal_resume_gc_inspect_errors_total";
#[derive(Debug, Clone, Copy)]
pub(crate) struct ResumeGcConfig {
/// Maximum number of directory entries considered in one disk pass.
pub(crate) max_entries: usize,
/// Maximum number of bytes read in one disk pass.
pub(crate) max_bytes: usize,
}
impl Default for ResumeGcConfig {
fn default() -> Self {
Self {
max_entries: DEFAULT_ENTRY_BUDGET,
max_bytes: DEFAULT_BYTE_BUDGET,
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ResumeGcReport {
/// Directory entries visited (including malformed entries).
pub(crate) inspected: usize,
pub(crate) active_skipped: usize,
pub(crate) orphaned: usize,
/// Records that must be handed to #1927's quarantine owner.
pub(crate) quarantine_required: usize,
pub(crate) generation_skipped: usize,
pub(crate) clock_skew: usize,
pub(crate) read_errors: usize,
pub(crate) retained: usize,
/// True while #1927's durable claim/quarantine capability is unavailable.
pub(crate) destructive_disabled: bool,
pub(crate) budget_exhausted: bool,
}
#[derive(Debug, Default)]
pub(crate) struct ResumeGc {
config: ResumeGcConfig,
/// Alternate the first namespace so a full ordinary page cannot starve
/// replacement recovery when the list API has no continuation token.
recovery_first: bool,
}
impl ResumeGc {
#[cfg(test)]
fn with_config(config: ResumeGcConfig) -> Self {
Self {
config,
recovery_first: false,
}
}
/// Inspect one bounded page from each resume namespace.
///
/// The caller owns scheduling and cancellation. A malformed or unreadable
/// artifact is reported and retained so a later pass can retry it; no
/// individual artifact error aborts the rest of the bounded page.
pub(crate) async fn inspect_disk(&mut self, disk: &DiskStore) -> Result<ResumeGcReport> {
let mut report = ResumeGcReport {
destructive_disabled: true,
..ResumeGcReport::default()
};
if self.config.max_entries == 0 || self.config.max_bytes == 0 {
report.budget_exhausted = true;
return Ok(report);
}
let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs();
let mut bytes_read = 0usize;
let recovery_first = self.recovery_first;
self.recovery_first = !self.recovery_first;
if recovery_first {
inspect_namespace(self.config, disk, &replacement_prefix(), true, now, &mut bytes_read, &mut report).await?;
if !report.budget_exhausted {
inspect_namespace(self.config, disk, BUCKET_META_PREFIX, false, now, &mut bytes_read, &mut report).await?;
}
} else {
inspect_namespace(self.config, disk, BUCKET_META_PREFIX, false, now, &mut bytes_read, &mut report).await?;
if !report.budget_exhausted {
inspect_namespace(self.config, disk, &replacement_prefix(), true, now, &mut bytes_read, &mut report).await?;
}
}
counter!(GC_METRIC).increment(u64::try_from(report.inspected).unwrap_or(u64::MAX));
counter!(GC_ERROR_METRIC).increment(u64::try_from(report.read_errors).unwrap_or(u64::MAX));
Ok(report)
}
}
#[derive(Debug, Default, Clone, Copy)]
struct ArtifactSet {
state: bool,
checkpoint: bool,
progress: bool,
replacement_intent: bool,
proof: bool,
seal: bool,
legacy_marker: bool,
temporary: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ArtifactKind {
State,
Checkpoint,
Progress,
ReplacementIntent,
Proof,
Seal,
LegacyMarker,
}
impl ArtifactSet {
fn add(&mut self, kind: ArtifactKind, temporary: bool) {
self.temporary |= temporary;
match kind {
ArtifactKind::State => self.state = true,
ArtifactKind::Checkpoint => self.checkpoint = true,
ArtifactKind::Progress => self.progress = true,
ArtifactKind::ReplacementIntent => self.replacement_intent = true,
ArtifactKind::Proof => self.proof = true,
ArtifactKind::Seal => self.seal = true,
ArtifactKind::LegacyMarker => self.legacy_marker = true,
}
}
}
#[derive(Debug, Clone, Copy)]
struct InspectOptions {
max_bytes: usize,
now: u64,
}
struct InspectProgress<'a> {
bytes_read: &'a mut usize,
report: &'a mut ResumeGcReport,
}
fn replacement_prefix() -> String {
super::replacement_recovery_dir().to_string_lossy().into_owned()
}
async fn inspect_namespace(
config: ResumeGcConfig,
disk: &DiskStore,
prefix: &str,
replacement: bool,
now: u64,
bytes_read: &mut usize,
report: &mut ResumeGcReport,
) -> Result<()> {
let remaining = config.max_entries.saturating_sub(report.inspected);
if remaining == 0 {
report.budget_exhausted = true;
return Ok(());
}
let count = i32::try_from(remaining).unwrap_or(i32::MAX);
let mut entries = match EcstoreDiskAPI::list_dir(disk.as_ref(), "", RUSTFS_META_BUCKET, prefix, count).await {
Ok(entries) => entries,
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return Ok(()),
Err(error) => return Err(error.into()),
};
entries.sort_unstable();
let mut artifacts = BTreeMap::<String, ArtifactSet>::new();
for entry in entries {
if report.inspected >= config.max_entries {
report.budget_exhausted = true;
break;
}
report.inspected += 1;
let Some((task_id, kind, temporary)) = artifact_name(&entry, replacement) else {
report.quarantine_required += 1;
report.retained += 1;
continue;
};
artifacts.entry(task_id).or_default().add(kind, temporary);
}
if report.inspected >= config.max_entries {
report.budget_exhausted = true;
}
for (task_id, artifacts) in artifacts {
if *bytes_read >= config.max_bytes {
report.budget_exhausted = true;
break;
}
let options = InspectOptions {
max_bytes: config.max_bytes,
now,
};
let mut progress = InspectProgress { bytes_read, report };
inspect_task(options, disk, prefix, replacement, &task_id, artifacts, &mut progress).await?;
}
Ok(())
}
async fn inspect_task(
options: InspectOptions,
disk: &DiskStore,
prefix: &str,
replacement: bool,
task_id: &str,
artifacts: ArtifactSet,
progress: &mut InspectProgress<'_>,
) -> Result<()> {
let legacy_replacement = !replacement && !artifacts.state && artifacts.replacement_intent;
let state_suffix = if replacement || legacy_replacement {
REPLACEMENT_INTENT_FILE
} else {
RESUME_STATE_FILE
};
let state_path = artifact_path(prefix, task_id, state_suffix)?;
let state = match read_bounded(disk, &state_path, options.max_bytes, progress.bytes_read).await {
ReadOutcome::Missing => {
progress.report.orphaned += 1;
progress.report.retained += 1;
return Ok(());
}
ReadOutcome::TooLarge => {
progress.report.quarantine_required += 1;
progress.report.retained += 1;
progress.report.budget_exhausted = true;
return Ok(());
}
ReadOutcome::Error => {
progress.report.read_errors += 1;
progress.report.retained += 1;
return Ok(());
}
ReadOutcome::Bytes(bytes) => bytes,
};
let parsed: ResumeState = match serde_json::from_slice(&state) {
Ok(state) => state,
Err(_) => {
progress.report.quarantine_required += 1;
progress.report.retained += 1;
return Ok(());
}
};
if parsed.schema_version > super::CURRENT_RESUME_SCHEMA || parsed.task_id != task_id {
progress.report.quarantine_required += 1;
progress.report.retained += 1;
return Ok(());
}
if persistent_age_seconds(options.now, parsed.last_update).is_none() {
progress.report.clock_skew += 1;
progress.report.retained += 1;
return Ok(());
}
if let Some(generation) = parsed.replacement_generation.as_deref()
&& !claim_generation_matches(Some(generation), Some(task_id))
{
progress.report.generation_skipped += 1;
progress.report.retained += 1;
return Ok(());
}
if !replacement && artifacts.checkpoint {
let checkpoint_path = artifact_path(prefix, task_id, RESUME_CHECKPOINT_FILE)?;
match read_bounded(disk, &checkpoint_path, options.max_bytes, progress.bytes_read).await {
ReadOutcome::Bytes(bytes) => match serde_json::from_slice::<ResumeCheckpoint>(&bytes) {
Ok(checkpoint) if checkpoint.schema_version <= CURRENT_CHECKPOINT_SCHEMA && checkpoint.task_id == task_id => {}
_ => {
progress.report.quarantine_required += 1;
progress.report.retained += 1;
}
},
ReadOutcome::Missing => {
progress.report.orphaned += 1;
progress.report.retained += 1;
}
ReadOutcome::TooLarge => {
progress.report.quarantine_required += 1;
progress.report.retained += 1;
progress.report.budget_exhausted = true;
}
ReadOutcome::Error => {
progress.report.read_errors += 1;
progress.report.retained += 1;
}
}
}
if artifacts.state && artifacts.replacement_intent {
// A task cannot have two authoritative state records in one namespace;
// preserve both until the durable owner can resolve the generation.
progress.report.quarantine_required += 1;
}
if !parsed.completed {
progress.report.active_skipped += 1;
}
// The state and all associated evidence remain recoverable until #1927
// supplies a common generation/CAS transition and quarantine owner.
progress.report.retained += 1;
Ok(())
}
enum ReadOutcome {
Bytes(Vec<u8>),
Missing,
TooLarge,
Error,
}
async fn read_bounded(disk: &DiskStore, path: &str, max_bytes: usize, bytes_read: &mut usize) -> ReadOutcome {
let remaining = max_bytes.saturating_sub(*bytes_read);
if remaining == 0 {
return ReadOutcome::TooLarge;
}
let read_len = remaining.saturating_add(1);
let reader = match EcstoreDiskAPI::read_file(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
Ok(reader) => reader,
Err(DiskError::FileNotFound | DiskError::VolumeNotFound) => return ReadOutcome::Missing,
Err(_) => return ReadOutcome::Error,
};
let mut bytes = Vec::with_capacity(read_len.min(64 * 1024));
let Ok(read_len) = u64::try_from(read_len) else {
return ReadOutcome::TooLarge;
};
if reader.take(read_len).read_to_end(&mut bytes).await.is_err() {
return ReadOutcome::Error;
}
if bytes.len() > remaining {
*bytes_read = max_bytes;
return ReadOutcome::TooLarge;
}
*bytes_read = bytes_read.saturating_add(bytes.len());
ReadOutcome::Bytes(bytes)
}
fn artifact_path(prefix: &str, task_id: &str, suffix: &str) -> Result<String> {
if super::validate_resume_task_id(task_id).is_err() {
return Err(Error::other("invalid resume task id"));
}
Path::new(prefix)
.join(format!("{task_id}_{suffix}"))
.to_str()
.map(str::to_owned)
.ok_or_else(|| Error::other("invalid resume artifact path"))
}
/// Parse one directory entry without ever accepting a path component supplied
/// by a client. DiskAPI filters symlinks, but this check also protects remote
/// implementations and future mutating callers from traversal/reparse names.
fn artifact_name(entry: &str, replacement: bool) -> Option<(String, ArtifactKind, bool)> {
let path = Path::new(entry);
if entry.is_empty() || path.components().count() != 1 || !matches!(path.components().next(), Some(Component::Normal(_))) {
return None;
}
let (stem, temporary) = entry
.strip_suffix(".tmp")
.map(|stem| (stem, true))
.or_else(|| entry.strip_suffix(".bak").map(|stem| (stem, true)))
.unwrap_or((entry, false));
let suffixes: &[(&str, ArtifactKind)] = if replacement {
&[
(REPLACEMENT_INTENT_FILE, ArtifactKind::ReplacementIntent),
(REPLACEMENT_COMPLETION_PROOF_FILE, ArtifactKind::Proof),
(REPLACEMENT_INTENT_SEAL_FILE, ArtifactKind::Seal),
]
} else {
&[
(RESUME_STATE_FILE, ArtifactKind::State),
(RESUME_CHECKPOINT_FILE, ArtifactKind::Checkpoint),
(RESUME_PROGRESS_FILE, ArtifactKind::Progress),
(LEGACY_REPLACEMENT_RECOVERY_MARKER_FILE, ArtifactKind::LegacyMarker),
(REPLACEMENT_INTENT_FILE, ArtifactKind::ReplacementIntent),
(REPLACEMENT_COMPLETION_PROOF_FILE, ArtifactKind::Proof),
(REPLACEMENT_INTENT_SEAL_FILE, ArtifactKind::Seal),
]
};
suffixes.iter().find_map(|(suffix, kind)| {
stem.strip_suffix(&format!("_{suffix}"))
.filter(|task_id| super::validate_resume_task_id(task_id).is_ok())
.map(|task_id| (task_id.to_string(), *kind, temporary))
})
}
fn persistent_age_seconds(now: u64, updated: u64) -> Option<u64> {
now.checked_sub(updated)
}
fn claim_generation_matches(observed: Option<&str>, expected: Option<&str>) -> bool {
expected.is_none() || observed == expected
}
#[cfg(test)]
mod tests {
use super::*;
use crate::heal::{DiskOption, Endpoint, new_disk};
use tempfile::TempDir;
use uuid::Uuid;
async fn test_disk() -> (TempDir, DiskStore) {
let temp = TempDir::new().expect("test disk directory");
let endpoint = Endpoint::try_from(temp.path().to_string_lossy().as_ref()).expect("test endpoint");
let disk = new_disk(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
)
.await
.expect("test disk");
match disk.make_volume(RUSTFS_META_BUCKET).await {
Ok(()) | Err(DiskError::VolumeExists) => {}
Err(error) => panic!("metadata volume: {error}"),
}
match disk.make_volume(&format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}")).await {
Ok(()) | Err(DiskError::VolumeExists) => {}
Err(error) => panic!("resume volume: {error}"),
}
(temp, disk)
}
async fn write_state(disk: &DiskStore, state: &ResumeState) {
let path = format!("{BUCKET_META_PREFIX}/{}_{}", state.task_id, RESUME_STATE_FILE);
disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(state).unwrap().into())
.await
.expect("resume state");
}
async fn write_replacement_state(disk: &DiskStore, state: &ResumeState) {
let volume = format!("{RUSTFS_META_BUCKET}/{BUCKET_META_PREFIX}/ahm-replacement");
match disk.make_volume(&volume).await {
Ok(()) | Err(DiskError::VolumeExists) => {}
Err(error) => panic!("replacement volume: {error}"),
}
let path = format!("{}/{}_{}", replacement_prefix(), state.task_id, REPLACEMENT_INTENT_FILE);
disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(state).unwrap().into())
.await
.expect("replacement state");
}
#[tokio::test]
async fn production_gc_does_not_delete_claimed_resume_state() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
write_state(&disk, &ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![])).await;
let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
assert_eq!(report.active_skipped, 1);
assert!(
disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
.await
.is_ok()
);
}
#[tokio::test]
async fn production_gc_generation_mismatch_is_skip() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
state.replacement_generation = Some(Uuid::new_v4().to_string());
write_state(&disk, &state).await;
assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().generation_skipped, 1);
}
#[cfg(unix)]
#[tokio::test]
async fn production_gc_rejects_symlink_or_outside_prefix() {
let (_temp, disk) = test_disk().await;
let id = Uuid::new_v4().to_string();
let root = EcstoreDiskAPI::path(disk.as_ref());
let outside = root.join("outside-resume-state");
std::fs::write(&outside, b"must remain").expect("outside fixture");
let symlink = root
.join(RUSTFS_META_BUCKET)
.join(BUCKET_META_PREFIX)
.join(format!("{id}_{RESUME_STATE_FILE}"));
std::os::unix::fs::symlink(&outside, &symlink).expect("symlink fixture");
let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
assert_eq!(report.inspected, 0, "symlinks are not eligible artifacts");
assert!(outside.exists());
assert!(artifact_name(&format!("{id}_{RESUME_STATE_FILE}"), false).is_some());
assert!(artifact_name(&format!("../{id}_{RESUME_STATE_FILE}"), false).is_none());
assert!(artifact_name(&format!("{id}/link_{RESUME_STATE_FILE}"), false).is_none());
}
#[cfg(not(unix))]
#[test]
fn production_gc_rejects_symlink_or_outside_prefix() {
let id = Uuid::new_v4().to_string();
assert!(artifact_name(&format!("{id}_{RESUME_STATE_FILE}"), false).is_some());
assert!(artifact_name(&format!("../{id}_{RESUME_STATE_FILE}"), false).is_none());
assert!(artifact_name(&format!("{id}/link_{RESUME_STATE_FILE}"), false).is_none());
}
#[tokio::test]
async fn production_gc_delete_failure_leaves_recoverable_state() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
state.mark_completed();
write_state(&disk, &state).await;
ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
assert!(
disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
.await
.is_ok()
);
}
#[tokio::test]
async fn production_gc_handles_clock_skew_and_restart() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let mut state = ResumeState::new(task_id, "set".into(), "disk".into(), vec![]);
state.last_update = u64::MAX;
write_state(&disk, &state).await;
assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().clock_skew, 1);
assert!(persistent_age_seconds(1, 2).is_none());
}
#[tokio::test]
async fn production_gc_100k_states_respects_budget() {
let (_temp, disk) = test_disk().await;
for _ in 0..8 {
let state = ResumeState::new(Uuid::new_v4().to_string(), "set".into(), "disk".into(), vec![]);
write_state(&disk, &state).await;
}
let config = ResumeGcConfig {
max_entries: 2,
max_bytes: usize::MAX,
};
let report = ResumeGc::with_config(config).inspect_disk(&disk).await.unwrap();
assert!(report.inspected <= 2);
assert!(report.budget_exhausted);
}
#[tokio::test]
async fn production_gc_recovery_namespace_is_not_starved() {
let (_temp, disk) = test_disk().await;
let ordinary = ResumeState::new(Uuid::new_v4().to_string(), "set".into(), "disk".into(), vec![]);
write_state(&disk, &ordinary).await;
let replacement_id = Uuid::new_v4().to_string();
let mut replacement = ResumeState::new(replacement_id, "set".into(), "disk".into(), vec![]);
replacement.replacement_generation = Some(replacement.task_id.clone());
write_replacement_state(&disk, &replacement).await;
let mut gc = ResumeGc::with_config(ResumeGcConfig {
max_entries: 1,
max_bytes: usize::MAX,
});
assert_eq!(gc.inspect_disk(&disk).await.unwrap().inspected, 1);
let second = gc.inspect_disk(&disk).await.unwrap();
assert_eq!(second.inspected, 1, "the next bounded pass must start at recovery");
assert_eq!(second.active_skipped, 1);
}
#[tokio::test]
async fn production_gc_pairs_orphan_checkpoint_and_resume() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let checkpoint = ResumeCheckpoint::new(task_id.clone());
let path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_CHECKPOINT_FILE}");
disk.write_all(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&checkpoint).unwrap().into())
.await
.expect("checkpoint");
assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().orphaned, 1);
}
#[tokio::test]
async fn production_gc_does_not_delete_slow_active_task() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let mut state = ResumeState::new(task_id, "set".into(), "disk".into(), vec![]);
state.last_update = 1;
write_state(&disk, &state).await;
assert_eq!(ResumeGc::default().inspect_disk(&disk).await.unwrap().active_skipped, 1);
}
#[tokio::test]
async fn production_gc_disables_on_mixed_version_capability() {
assert!(claim_generation_matches(None, None));
assert!(!claim_generation_matches(Some("new"), Some("old")));
// No #1927 capability means this implementation has no delete path.
assert!(ResumeGcConfig::default().max_entries > 0);
let (_temp, disk) = test_disk().await;
let report = ResumeGc::default().inspect_disk(&disk).await.expect("inspect");
assert!(report.destructive_disabled);
}
#[tokio::test]
async fn production_gc_future_schema_is_not_mtime_deleted() {
let (_temp, disk) = test_disk().await;
let task_id = Uuid::new_v4().to_string();
let mut state = ResumeState::new(task_id.clone(), "set".into(), "disk".into(), vec![]);
state.schema_version = super::super::CURRENT_RESUME_SCHEMA + 1;
write_state(&disk, &state).await;
let report = ResumeGc::default().inspect_disk(&disk).await.unwrap();
assert_eq!(report.quarantine_required, 1);
assert!(
disk.read_all(RUSTFS_META_BUCKET, &format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}"))
.await
.is_ok()
);
}
#[tokio::test]
async fn production_gc_quarantine_cleanup_is_bounded() {
let (_temp, disk) = test_disk().await;
for _ in 0..4 {
let task_id = Uuid::new_v4().to_string();
let path = format!("{BUCKET_META_PREFIX}/{task_id}_{RESUME_STATE_FILE}");
disk.write_all(RUSTFS_META_BUCKET, &path, b"corrupt".to_vec().into())
.await
.expect("corrupt state");
}
let report = ResumeGc::with_config(ResumeGcConfig {
max_entries: 2,
max_bytes: 1024,
})
.inspect_disk(&disk)
.await
.expect("inspect");
assert!(report.quarantine_required <= 2);
assert!(report.budget_exhausted);
}
}
-3
View File
@@ -43,9 +43,6 @@ 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
@@ -0,0 +1,149 @@
# CI gate matrix
This file is the source of truth for which validation runs on each event, its
configured wall-clock budget, and whether it can block a merge. Test taxonomy,
naming, and nextest serialization rules remain in [README.md](README.md); e2e
membership and counts remain in
[e2e-suite-inventory.md](e2e-suite-inventory.md).
The distinction between **required** and **report-only** is load-bearing:
a failing job blocks a merge only when its exact check name is present in the
live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
does not make a job required by itself.
## Required merge checks
The live `main` ruleset (`6436880`) currently requires exactly these contexts:
| Required context | Producer | Validation |
|---|---|---|
| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
For pull requests limited to the paths excluded by the main CI workflow,
`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
`Test and Lint` under the same names. It runs the real quick checks and the
planning-document guard; it does not claim that Rust compilation or runtime
tests ran. Despite the workflow name, these paths also include selected deploy,
workflow, and lock files.
Verify the live rule rather than trusting this snapshot before changing merge
policy:
```bash
gh api repos/rustfs/rustfs/rulesets/6436880 \
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
```
The ruleset currently has `strict_required_status_checks_policy=false`.
`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
them, but `End-to-End Tests (full merge gate)` is not currently a required
context. Therefore the repository is prepared to test a merge-queue SHA, but
the workflow alone does not prove that every merge passed that lane.
## Pull request and merge matrix
Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
means the result is visible and actionable but is not in the live required
context list.
| Event | Validation | Budget | Merge status | Reproduction |
|---|---|---:|---|---|
| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
instead of adding a second ad-hoc selector. Before a profile runs,
`scripts/check_test_wiring.py` compares its exact membership to the committed
digest so a silent test drop fails closed.
## Scheduled and manual validation
Scheduled lanes are independent fault domains. They do not block a pull
request, but their workflow-local gate can fail the run and scheduled failures
are routed to the shared failure-issue action. The scheduled-validation
watchdog and freshness workflow separately detect incomplete runs and missing
schedules.
| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
|---|---|---:|---|---|
| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
runs are debugging evidence and intentionally do not open scheduled-failure
issues. A manual performance run may explicitly allow a known regression; that
override must not be treated as an ordinary passing baseline.
## Release validation
Release validation is post-merge and tag-driven; it does not substitute for a
pull-request gate.
| Event | Validation | Budget | Result |
|---|---|---:|---|
| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
are backfill/debug paths and do not prove the automatic `workflow_run` chain.
## Evidence requirements
A green check is useful only when it proves the intended behavior ran:
- Record the exact commit SHA and run URL.
- Separate product failure from runner prerequisites, service readiness, and
cancellation. Repair the precondition, then rerun the exact workload.
- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
server logs where the workflow provides them.
- For a bug fix or a new fault checker, provide sensitivity evidence: the old
behavior or an intentional mutation must fail the new oracle, and the fixed
behavior must pass it.
- Never promote a report-only lane to required from one green run. Require at
least 14 days and 30 representative pull requests with at least 99% complete
execution, then update the ruleset and this table together.
## Change checklist
Update this file in the same pull request when any of these change:
- workflow triggers, job names, timeouts, or nextest profile ownership;
- required status contexts or strict/merge-queue policy;
- scheduled cadence, alert routing, artifact contract, or local reproduction;
- report-only versus gating semantics.
Do not copy per-module test counts here. Update
[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
digest instead.
+2 -2
View File
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
mimalloc = { workspace = true }
rustfs-mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
libmimalloc-sys.workspace = true
rustfs-mimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+1 -7
View File
@@ -369,14 +369,8 @@ 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> {
// 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);
}
rustfs_mimalloc::MiMalloc::collect(force);
Ok(())
}
+10 -8
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 { mimalloc::MiMalloc.alloc(layout) }
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
}
}
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,8 +71,9 @@ 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 { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
assert!(unsafe { heap.contains(allocation.as_ptr()) });
}
#[test]
@@ -85,12 +86,13 @@ 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 { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -102,7 +104,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
assert!(unsafe { heap.contains(grown_ptr) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+19 -36
View File
@@ -17,10 +17,7 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split,
};
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;
@@ -231,7 +228,18 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
#[cfg(any(test, not(target_os = "windows")))]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
let json = rustfs_mimalloc::MiMalloc::stats_json();
if json.is_empty() {
return None;
}
let observation = parse_mimalloc_stats_json(&json)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -242,7 +250,6 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
@@ -254,7 +261,6 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
@@ -271,12 +277,10 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
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()
@@ -285,7 +289,6 @@ 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"];
@@ -312,33 +315,6 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
let observation = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
libmimalloc_sys::mi_free(stats_ptr.cast());
observation?
};
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -566,6 +542,13 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[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);
+12 -81
View File
@@ -219,7 +219,7 @@ def check_s3_tests_runner(root: Path) -> list[str]:
return []
def profile_selection_entries(root: Path, profile: str) -> tuple[Path, list[str], dict[str, str]]:
def profile_selection(root: Path, profile: str) -> str:
if not re.fullmatch(r"e2e-[a-z0-9-]+", profile):
raise ValueError(f"invalid e2e profile name: {profile}")
path = root / f".config/{profile}-selection.txt"
@@ -227,11 +227,6 @@ def profile_selection_entries(root: Path, profile: str) -> tuple[Path, list[str]
values = dict(line.split("=", 1) for line in lines if "=" in line)
if len(values) != len(lines) or any(not re.fullmatch(r"sha256(?:-[a-z0-9]+)?", key) for key in values):
raise ValueError(f"{path.relative_to(root).as_posix()}: invalid sha256 entry")
return path, lines, values
def profile_selection(root: Path, profile: str) -> str:
path, _, values = profile_selection_entries(root, profile)
key = f"sha256-{sys.platform}"
digest = values.get(key, values.get("sha256", ""))
if not re.fullmatch(r"[0-9a-f]{64}", digest):
@@ -239,29 +234,6 @@ def profile_selection(root: Path, profile: str) -> str:
return digest
def profile_listing_digest(listing: Path) -> tuple[int, str]:
data = json.loads(listing.read_text())
selected = sorted(
f"{suite_id}::{test_name}"
for suite_id, suite in data["rust-suites"].items()
for test_name, testcase in suite["testcases"].items()
if testcase.get("filter-match", {}).get("status") == "matches"
)
return len(selected), hashlib.sha256(("\n".join(selected) + "\n").encode()).hexdigest()
def update_profile_selection(root: Path, profile: str, listing: Path, platform: str) -> tuple[int, str, str]:
if not re.fullmatch(r"[a-z0-9]+", platform):
raise ValueError(f"invalid platform name: {platform}")
path, lines, values = profile_selection_entries(root, profile)
key = "sha256" if "sha256" in values else f"sha256-{platform}"
if key not in values:
raise ValueError(f"{path.relative_to(root).as_posix()}: missing {key} entry")
count, digest = profile_listing_digest(listing)
path.write_text("\n".join(f"{key}={digest}" if line.startswith(f"{key}=") else line for line in lines) + "\n")
return count, digest, key
def check_profile_definitions(root: Path) -> list[str]:
config = tomllib.loads((root / ".config/nextest.toml").read_text())
profiles = {
@@ -575,15 +547,22 @@ def check_scheduled_alerts(root: Path) -> list[str]:
def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
try:
expected_digest = profile_selection(root, profile)
count, digest = profile_listing_digest(listing)
data = json.loads(listing.read_text())
selected = sorted(
f"{suite_id}::{test_name}"
for suite_id, suite in data["rust-suites"].items()
for test_name, testcase in suite["testcases"].items()
if testcase.get("filter-match", {}).get("status") == "matches"
)
digest = hashlib.sha256(("\n".join(selected) + "\n").encode()).hexdigest()
except (FileNotFoundError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
return [f"cannot read {profile} nextest listing: {error}"]
if digest != expected_digest:
return [
f"{profile} selection changed: count={count} sha256={digest}; "
f"{profile} selection changed: count={len(selected)} sha256={digest}; "
f"expected sha256={expected_digest}"
]
print(f"{profile} selection OK: {count} tests, sha256={digest}")
print(f"{profile} selection OK: {len(selected)} tests, sha256={digest}")
return []
@@ -728,45 +707,6 @@ class SelfTests(unittest.TestCase):
with mock.patch.object(sys, "platform", "linux"):
self.assertEqual(len(check_profile_listing(root, "e2e-full", listing)), 1)
def test_update_profile_selection_changes_only_requested_platform(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / ".config").mkdir()
darwin_digest = "a" * 64
(root / ".config/e2e-full-selection.txt").write_text(
f"sha256-darwin={darwin_digest}\nsha256-linux={'b' * 64}\n"
)
listing = root / "listing.json"
listing.write_text(
json.dumps(
{
"rust-suites": {
"suite": {
"testcases": {"linux": {"filter-match": {"status": "matches"}}}
}
}
}
)
)
count, digest, key = update_profile_selection(root, "e2e-full", listing, "linux")
self.assertEqual(count, 1)
self.assertEqual(key, "sha256-linux")
self.assertEqual(
(root / ".config/e2e-full-selection.txt").read_text(),
f"sha256-darwin={darwin_digest}\nsha256-linux={digest}\n",
)
with mock.patch.object(sys, "platform", "linux"):
self.assertEqual(check_profile_listing(root, "e2e-full", listing), [])
selection = root / ".config/e2e-smoke-selection.txt"
selection.write_text(f"sha256={'a' * 64}\n")
count, digest, key = update_profile_selection(root, "e2e-smoke", listing, "linux")
self.assertEqual((count, key), (1, "sha256"))
self.assertEqual(selection.read_text(), f"sha256={digest}\n")
def test_scheduled_alerts_require_completion_watchdog(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
@@ -1044,18 +984,9 @@ def main() -> int:
print(f"ERROR: {error}", file=sys.stderr)
return 1
return 0
if len(sys.argv) == 5 and sys.argv[1] == "--update-profile":
try:
count, digest, key = update_profile_selection(ROOT, sys.argv[2], Path(sys.argv[3]), sys.argv[4])
except (FileNotFoundError, KeyError, TypeError, ValueError, json.JSONDecodeError) as error:
print(f"ERROR: cannot update {sys.argv[2]} selection: {error}", file=sys.stderr)
return 1
print(f"Updated .config/{sys.argv[2]}-selection.txt: count={count} {key}={digest}")
return 0
if sys.argv[1:]:
print(
"usage: check_test_wiring.py [--self-test | --check-profile PROFILE LISTING | "
"--update-profile PROFILE LISTING PLATFORM]",
"usage: check_test_wiring.py [--self-test | --check-profile PROFILE LISTING]",
file=sys.stderr,
)
return 2