Compare commits

...

10 Commits

Author SHA1 Message Date
cxymds 1afaf4491e fix(ecstore): bound decommission capacity gate waits (#8044)
* fix(ecstore): bound decommission capacity gate waits

* fix(ecstore): keep rebalance meta save lock errors retryable

The rebalance metadata retry policy only retries when the source chain still
carries the typed lock error, but resolve_rebalance_meta_save_result collapsed
every failure into a plain string. A transient rebalance.bin write-lock timeout
therefore bypassed retry_rebalance_metadata_access entirely and surfaced as a
hard failure of the periodic and terminal rebalance metadata saves.

Wrap the failure with the data-movement stage context instead, which keeps the
original lock error reachable through rebalance_error_source so the existing
transient lock policy applies. The rendered message is unchanged.

Add a regression test that feeds the wrapped meta save lock timeout through the
retry helper and asserts the retry engages.

* chore(ecstore): refresh error format ratchet baseline

Removing the string-wrapped meta save failure drops one `::other(format!`
call site in the rebalance worker, so the shrink-only baseline must be
regenerated in the same change.
2026-09-21 20:56:26 +08:00
hector bdea44f332 fix(ci): align FT evidence contract with scenario E (44 cases) (#8052)
The fault-tolerance suite now registers and runs scenario E
(rustfs/auto-testing#100): --all covers A, B, C, C2, D, E with 38 + 6
= 44 expected cases. The chain-evidence assertion still hardcoded
[A, B, C, C2, D] and 38 everywhere, which would fail every green FT
run's evidence validation once E is registered.

Bump the scenario list to include E and all count assertions from 38
to 44; step name updated to match. No other lanes reference 38.
2026-09-21 20:13:56 +08:00
cxymds 8ba3309624 fix(rebalance): retry fleet proof before start activation (#8049)
Retry the transient missing fleet capability proof in both the admin start flow and node RPC activation. Return a retryable 503 with Retry-After after rollback when readiness does not converge.
2026-09-21 11:27:58 +00:00
cxymds cd2fa50e50 fix: rename admin metrics endpoint to realtime (#8046) 2026-09-21 19:23:58 +08:00
cxymds 077b961232 fix(obs): default span events to none and gate span list by level (#8047)
JSON log layers synthesized span lifecycle events for every span and
embedded the full ancestor span list on every record. With INFO spans on
object hot paths that produced multi-megabyte lines, which journald
truncated at 48 KiB: the payload was silently dropped and the volume
itself became the dominant load.

Add RUSTFS_OBS_LOG_SPAN_EVENTS (none|close|full) with a default of none,
and keep with_span_list enabled only for verbose (debug/trace) levels so
the ancestor chain stops being re-serialized per record. The request_id
promotion path reads the span scope, not the span list, so it is
unaffected.

Refs rustfs/backlog#2642
2026-09-21 19:15:39 +08:00
cxymds e1bc731331 fix(lifecycle): reject empty expiration actions (#8045)
* fix(lifecycle): reject empty expiration actions

* style: apply rustfmt to lifecycle validation
2026-09-21 15:43:07 +08:00
cxymds bb43ef861a fix(decommission): use owned data for source capacity (#8038) 2026-09-21 15:29:30 +08:00
Chris 4047ed6c8d fix(heal): restore bucket metadata before replacement completion (#8042) 2026-09-21 12:18:06 +08:00
hector ae6bbaef27 fix(ci): give the chain staleness probe a token that can read auto-testing (#8041)
resolve_functional_candidate.py probes rustfs/auto-testing (private) to
age the pinned functional-script revision and fall back to main HEAD
after 24h. The prepare step passed github.token, which cannot see the
private repo, so every nightly chain logged

  staleness probe failed (...exit status 1.); keeping pinned revision

and replayed the 09-14 harness. On the 09-20 nightly that harness died
on the dpkg conffile prompt in all 12 lanes (see rustfs/auto-testing#97)
because the --force-confold and other fixes never reached the chain.

Use PF_TESTING_GH_TOKEN - already required by the other steps in this
workflow - so the probe can actually run and the >24h fallback works.
2026-09-21 11:08:41 +08:00
cxymds a7c875b48f fix(heal): preserve remote corruption and reject incomplete deep scans (#8032)
* fix(heal): reject truncated encoded shards

* test(heal): cover truncated shards across integrity modes (#8035)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

* fix(heal): preserve remote corruption and reject incomplete deep scans

* test(admin): tolerate config lock timeout while polling

* fix(test): use admin storage error facade

---------

Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
2026-09-21 02:33:03 +00:00
41 changed files with 2060 additions and 208 deletions
@@ -149,7 +149,7 @@ jobs:
CHAIN_MANIFEST: ${{ inputs.chain_manifest }}
run: python3 scripts/prepare_functional_package.py prepare
- name: Run fault-tolerance scenarios (A, B, C, C2, D)
- name: Run fault-tolerance scenarios (A, B, C, C2, D, E)
timeout-minutes: 45
id: test
# Standalone case failures are reported by the backlog manager.
@@ -194,21 +194,21 @@ jobs:
jq -e --arg testing_sha "${AUTO_TESTING_SHA}" '
.schema_version == 1 and
.auto_testing_sha == $testing_sha and
.selected_scenarios == ["A", "B", "C", "C2", "D"] and
.selected_scenarios == ["A", "B", "C", "C2", "D", "E"] and
.complete == true and
.counts.expected == 38 and
.counts.seen == 38 and
.counts.expected == 44 and
.counts.seen == 44 and
.counts.harness_errors == 0 and
(.expected_cases | length) == 38 and
(.expected_cases | unique | length) == 38 and
(.seen_cases | length) == 38 and
(.seen_cases | unique | length) == 38 and
(.expected_cases | length) == 44 and
(.expected_cases | unique | length) == 44 and
(.seen_cases | length) == 44 and
(.seen_cases | unique | length) == 44 and
(.seen_cases | sort) == (.expected_cases | sort) and
(.missing_cases | length) == 0 and
(.harness_errors | length) == 0 and
(.cases | length) == 38 and
(.cases | length) == 44 and
([.cases[].case_id] | sort) == (.expected_cases | sort) and
([.cases[].case_id] | unique | length) == 38 and
([.cases[].case_id] | unique | length) == 44 and
all(.cases[]; .verdict == "pass" or .verdict == "known-divergence" or .verdict == "UNEXPECTED") and
.counts.unexpected == ([.cases[] | select(.verdict == "UNEXPECTED")] | length) and
.counts.known_divergence == ([.cases[] | select(.verdict == "known-divergence")] | length)
@@ -54,7 +54,10 @@ jobs:
- name: Resolve published candidate
id: candidate
env:
GH_TOKEN: ${{ github.token }}
# The staleness probe in resolve_functional_candidate.py reads the
# private rustfs/auto-testing repo; github.token cannot, and a probe
# failure silently keeps a stale testing_sha pinned for the chain.
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
BUILD_RUN_ID: ${{ inputs.build_run_id }}
BUILD_RUN_ATTEMPT: ${{ inputs.build_run_attempt }}
CHAIN_OUTPUT: ${{ runner.temp }}/chain-candidate.json
+11
View File
@@ -55,6 +55,13 @@ pub const ENV_OBS_LOG_FILENAME: &str = "RUSTFS_OBS_LOG_FILENAME";
pub const ENV_OBS_LOG_ROTATION_TIME: &str = "RUSTFS_OBS_LOG_ROTATION_TIME";
pub const ENV_OBS_LOG_KEEP_FILES: &str = "RUSTFS_OBS_LOG_KEEP_FILES";
/// Span lifecycle events emitted by the JSON log layers (`none`|`close`|`full`).
///
/// Defaults to `none`: a record per span lifecycle multiplies log volume on
/// object hot paths, and emitting the ancestor span list on every record
/// produced multi-megabyte lines that journald truncated (backlog#2642).
pub const ENV_OBS_LOG_SPAN_EVENTS: &str = "RUSTFS_OBS_LOG_SPAN_EVENTS";
/// Log cleanup related configurations
pub const ENV_OBS_LOG_MAX_TOTAL_SIZE_BYTES: &str = "RUSTFS_OBS_LOG_MAX_TOTAL_SIZE_BYTES";
pub const ENV_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES: &str = "RUSTFS_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES";
@@ -99,6 +106,8 @@ pub const DEFAULT_OBS_LOG_DRY_RUN: bool = false;
pub const DEFAULT_OBS_LOG_MATCH_MODE_PREFIX: &str = "prefix";
pub const DEFAULT_OBS_LOG_MATCH_MODE: &str = "suffix";
pub const DEFAULT_OBS_LOG_SPAN_EVENTS: &str = "none";
/// Default values for observability configuration
// ### Supported Environment Values
// - `production` - Secure file-only logging
@@ -141,6 +150,7 @@ mod tests {
assert_eq!(ENV_OBS_LOG_FILENAME, "RUSTFS_OBS_LOG_FILENAME");
assert_eq!(ENV_OBS_LOG_ROTATION_TIME, "RUSTFS_OBS_LOG_ROTATION_TIME");
assert_eq!(ENV_OBS_LOG_KEEP_FILES, "RUSTFS_OBS_LOG_KEEP_FILES");
assert_eq!(ENV_OBS_LOG_SPAN_EVENTS, "RUSTFS_OBS_LOG_SPAN_EVENTS");
assert_eq!(ENV_OBS_TRACES_EXPORT_ENABLED, "RUSTFS_OBS_TRACES_EXPORT_ENABLED");
assert_eq!(ENV_OBS_METRICS_EXPORT_ENABLED, "RUSTFS_OBS_METRICS_EXPORT_ENABLED");
assert_eq!(ENV_OBS_LOGS_EXPORT_ENABLED, "RUSTFS_OBS_LOGS_EXPORT_ENABLED");
@@ -178,6 +188,7 @@ mod tests {
assert_eq!(DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM_GZIP, "gzip");
assert_eq!(DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM_ZSTD, "zstd");
assert_eq!(DEFAULT_OBS_LOG_MATCH_MODE_PREFIX, "prefix");
assert_eq!(DEFAULT_OBS_LOG_SPAN_EVENTS, "none");
assert_eq!(DEFAULT_OBS_LOG_MATCH_MODE, "suffix");
assert_eq!(DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM, "zstd");
}
@@ -184,7 +184,7 @@ async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent()
let (status, metrics_body) = admin_request(
&node.url,
Method::GET,
"/rustfs/admin/v3/metrics?n=1&by-host=true&by-disk=true",
"/rustfs/admin/v3/realtime?n=1&by-host=true&by-disk=true",
None,
&dist.cluster.access_key,
&dist.cluster.secret_key,
@@ -257,7 +257,7 @@ async fn node_admin_body(dist: &DistCluster, node: usize, path: &str) -> TestRes
}
async fn http_put_counts(dist: &DistCluster, node: usize) -> TestResult<[u64; 2]> {
let body = node_admin_body(dist, node, "/rustfs/admin/v3/metrics?types=512&by-host=true&n=1").await?;
let body = node_admin_body(dist, node, "/rustfs/admin/v3/realtime?types=512&by-host=true&n=1").await?;
let sample: RealtimeMetrics = serde_json::from_str(body.lines().next().ok_or("empty HTTP metrics stream")?)?;
assert!(sample.errors.is_empty(), "HTTP metrics returned errors: {:?}", sample.errors);
let http = sample.aggregated.http.ok_or("HTTP metrics missing at WARN log level")?;
@@ -2141,7 +2141,7 @@ impl RemoteDisk {
let file_info_bin = encode_file_info_msgpack(fi)?;
let mut client = self.get_client().await?;
let mut request = Request::new(RenameDataRequest {
bucket_incarnation_id: crate::store::bucket_heal_scope(dst_volume)
bucket_incarnation_id: crate::store::bucket_heal_scope_for_object(dst_volume, dst_path)
.map(|scope| scope.incarnation.as_bytes().to_vec().into())
.unwrap_or_default(),
disk: self.endpoint.to_string(),
+356 -33
View File
@@ -41,7 +41,7 @@ use crate::config::com::{
};
use crate::data_movement;
use crate::data_movement::backpressure::{self, DataMovementOperation};
use crate::data_usage::DATA_USAGE_CACHE_NAME;
use crate::data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, load_data_usage_cache};
use crate::disk::error::DiskError;
use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
@@ -135,6 +135,7 @@ const DECOMMISSION_CAPACITY_RELEASE_FAILED: &str = "failed";
const DECOMMISSION_CAPACITY_RELEASE_COMPLETED: &str = "completed";
pub(crate) const DECOMMISSION_CAPACITY_TARGET_LOCK_PREFIX: &str = "decommission/capacity-target";
const DECOMMISSION_CAPACITY_TARGET_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250);
const DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS: usize = 12;
const DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX: &str = "target pool ";
const DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX: &str = " target capacity mutation gate is busy";
const METRIC_DECOMMISSION_CAPACITY_CONFLICTS_TOTAL: &str = "rustfs_decommission_capacity_conflicts_total";
@@ -158,6 +159,10 @@ const DECOMMISSION_COPY_RETRY_DELAY: std::time::Duration = std::time::Duration::
const DECOMMISSION_SOURCE_CHANGED_EXHAUSTION_LIMIT: usize = 100;
const DECOMMISSION_TERMINAL_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
const DECOMMISSION_CANCEL_TARGET_LOCK_MAX_ATTEMPTS: usize = 3;
fn decommission_capacity_target_gate_retry_exhausted(attempt: usize) -> bool {
attempt.saturating_add(1) >= DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS
}
const DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT: &str = "decommission/ilm-receipts";
const DECOMMISSION_DURABLE_ILM_MANIFEST_ROOT: &str = "decommission/ilm-manifests";
const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2";
@@ -925,6 +930,29 @@ fn capacity_target_physical_bytes(data_bytes: usize, layout: DecommissionErasure
Ok(capacity_mul_div_ceil(data_bytes, layout.width(), layout.data))
}
async fn decommission_owned_physical_usage(sets: &Sets, layout: DecommissionErasureLayout) -> HashMap<i32, usize> {
let mut usage = HashMap::with_capacity(sets.disk_set.len());
for (set_index, set) in sets.disk_set.iter().enumerate() {
let Ok(cache) = load_data_usage_cache(set.as_ref(), DATA_USAGE_CACHE_NAME).await else {
continue;
};
let data_usage = cache.dui(DATA_USAGE_ROOT, &[]);
if !data_usage.is_complete_bucket_usage_snapshot() {
continue;
}
let Ok(logical_bytes) = usize::try_from(data_usage.objects_total_size) else {
continue;
};
let Ok(set_index) = i32::try_from(set_index) else {
continue;
};
// The data-usage snapshot excludes unrelated files on a shared mount.
// Convert logical object bytes to the source set's physical EC footprint.
usage.insert(set_index, capacity_mul_div_ceil(logical_bytes, layout.width(), layout.data));
}
usage
}
fn worst_decommission_target_layout(targets: &[DecommissionPoolCapacityInfo]) -> Result<DecommissionErasureLayout> {
targets
.iter()
@@ -1861,6 +1889,132 @@ fn ensure_external_decommission_target_admission(
Err(Error::SlowDown)
}
fn grow_decommission_target_reservation(
meta: &mut PoolMeta,
capacity_infos: &[DecommissionPoolCapacityInfo],
source_pool_index: usize,
target_pool_index: usize,
expected_data_bytes: usize,
mutation_id: uuid::Uuid,
now: OffsetDateTime,
) -> Result<bool> {
let original = meta.clone();
let result = (|| {
let target_capacity = capacity_infos
.iter()
.find(|capacity| capacity.pool_index == target_pool_index)
.ok_or_else(|| {
decommission_capacity_blocked_error("target capacity snapshot is missing during reservation growth")
})?;
let pool_count = meta.pools.len();
let pool = meta
.pools
.get_mut(source_pool_index)
.ok_or_else(|| invalid_decommission_pool_index_error(pool_count, source_pool_index))?;
let reservation = pool
.decommission
.as_mut()
.and_then(|info| info.capacity_reservation.as_mut())
.filter(|reservation| reservation.active())
.ok_or_else(|| decommission_capacity_blocked_error("active reservation disappeared during reservation growth"))?;
let target_layout = reservation
.targets
.iter()
.find(|target| target.pool_index == target_pool_index)
.map(|target| target.layout)
.unwrap_or(target_capacity.layout);
let data_bytes = expected_data_bytes.max(1);
let required_target_bytes = capacity_target_physical_bytes(data_bytes, target_layout)?;
let required_peak = required_target_bytes.saturating_mul(1usize.saturating_add(reservation.temporary_copies));
let existing_target = reservation
.targets
.iter()
.find(|target| target.pool_index == target_pool_index);
if let Some(target) = existing_target
&& target.pending_physical_bytes > 0
&& target.pending_mutation_id != Some(mutation_id)
{
return Err(decommission_capacity_blocked_error(format!(
"source pool {source_pool_index} target pool {target_pool_index} has an unresolved target capacity intent"
)));
}
let existing_remaining = existing_target
.map(|target| target.remaining_reserved_physical_bytes(reservation.temporary_copies))
.unwrap_or_default();
if required_peak <= existing_remaining {
return Ok(false);
}
if reservation.source_data_equivalent_bytes > 0 {
return Err(decommission_capacity_blocked_error(format!(
"source pool {source_pool_index} reservation estimate is exhausted"
)));
}
if let Some(target) = existing_target
&& target_capacity.physical_free <= target.physical_free_at_reservation
{
return Err(decommission_capacity_blocked_error(format!(
"source pool {source_pool_index} target pool {target_pool_index} reservation is exhausted"
)));
}
let current_peak = reservation.remaining_peak_physical_bytes();
let additional_peak = required_peak.saturating_sub(existing_remaining);
let additional_predicted = additional_peak.div_ceil(1usize.saturating_add(reservation.temporary_copies));
let additional_source_data = capacity_source_data_equivalent(additional_predicted, target_layout)?;
reservation.source_data_equivalent_bytes = reservation
.source_data_equivalent_bytes
.saturating_add(additional_source_data);
reservation.source_physical_bytes = reservation
.source_physical_bytes
.saturating_add(capacity_target_physical_bytes(additional_source_data, reservation.source_layout)?);
reservation.predicted_physical_bytes = reservation
.predicted_physical_bytes
.saturating_add(additional_predicted)
.max(capacity_target_physical_bytes(reservation.source_data_equivalent_bytes, target_layout)?);
reservation.temporary_physical_bytes = reservation
.predicted_physical_bytes
.saturating_mul(reservation.temporary_copies);
reservation.peak_physical_bytes = reservation
.predicted_physical_bytes
.saturating_add(reservation.temporary_physical_bytes);
let allocation_growth = reservation.remaining_peak_physical_bytes().saturating_sub(current_peak);
if let Some(target) = reservation
.targets
.iter_mut()
.find(|target| target.pool_index == target_pool_index)
{
target.reserved_physical_bytes = target.reserved_physical_bytes.saturating_add(allocation_growth);
target.physical_total_at_reservation = target.physical_total_at_reservation.max(target_capacity.physical_total);
target.physical_free_at_reservation = target.physical_free_at_reservation.max(target_capacity.physical_free);
} else {
reservation.targets.push(DecommissionCapacityTarget {
pool_index: target_pool_index,
layout: target_layout,
physical_total_at_reservation: target_capacity.physical_total,
physical_free_at_reservation: target_capacity.physical_free,
reserved_physical_bytes: allocation_growth,
consumed_physical_bytes: 0,
observed_physical_bytes: 0,
inflight_physical_bytes: 0,
pending_physical_bytes: 0,
pending_mutation_id: None,
temporary_mutations: Vec::new(),
});
}
renew_decommission_capacity_reservation(reservation, now, true);
pool.last_update = pool.last_update.max(now);
Ok(true)
})();
if result.is_ok() {
if let Err(err) = ensure_decommission_capacity_reservations_available(meta, capacity_infos, "reservation growth") {
*meta = original;
return Err(err);
}
} else {
*meta = original;
}
result
}
fn ensure_decommission_target_owner_admission(
meta: &PoolMeta,
owner: DecommissionCapacityOwner,
@@ -11402,10 +11556,11 @@ impl ECStore {
}
})
.filter(|reservation| {
reservation
.targets
.iter()
.any(|target| target.pool_index == target_pool_index)
!temporary_release
|| reservation
.targets
.iter()
.any(|target| target.pool_index == target_pool_index)
})
.map(|reservation| (owner, reservation.model_version))
});
@@ -11464,10 +11619,11 @@ impl ECStore {
} else {
reservation.admits_owner(owner, OffsetDateTime::now_utc())
}) && reservation.model_version == model_version
&& reservation
.targets
.iter()
.any(|target| target.pool_index == target_pool_index)
&& (!temporary_release
|| reservation
.targets
.iter()
.any(|target| target.pool_index == target_pool_index))
})
.is_some();
if !owner_current {
@@ -11476,9 +11632,22 @@ impl ECStore {
));
}
let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?;
if !temporary_release {
let reservation_grew = if temporary_release {
false
} else {
let data_bytes = expected_data_bytes.unwrap_or(1).max(1);
let grew = grow_decommission_target_reservation(
&mut snapshot,
&capacity_infos,
source_pool_index,
target_pool_index,
data_bytes,
mutation_id,
OffsetDateTime::now_utc(),
)?;
ensure_decommission_capacity_reservations_available(&snapshot, &capacity_infos, "mutation")?;
}
grew
};
let target_layout = snapshot
.pools
.get(source_pool_index)
@@ -11544,7 +11713,7 @@ impl ECStore {
OffsetDateTime::now_utc(),
)?
};
if pending_added > 0 {
if reservation_grew || pending_added > 0 {
let outcome = snapshot
.save_no_lock_armed(self.pools.clone(), &mut save_guard, write_guard.lock_lost_signal(), &[source_pool_index])
.await?;
@@ -12386,7 +12555,11 @@ impl ECStore {
Ok(())
}
async fn get_decommission_pool_capacity_info(&self, idx: usize) -> Result<DecommissionPoolCapacityInfo> {
async fn get_decommission_pool_capacity_info(
&self,
idx: usize,
use_owned_usage: bool,
) -> Result<DecommissionPoolCapacityInfo> {
if let Some(sets) = self.pools.get(idx) {
let mut info = sets.storage_info_snapshot().await;
info.backend = StorageAdminApi::backend_info(self).await;
@@ -12418,8 +12591,13 @@ impl ECStore {
layout.data, layout.parity
)));
}
let owned_physical_used = if use_owned_usage {
Some(decommission_owned_physical_usage(sets, layout).await)
} else {
None
};
let (physical_total, physical_free, physical_used) =
decommission_physical_pool_capacity(&info.disks, idx, layout, space);
decommission_physical_pool_capacity(&info.disks, idx, layout, space, owned_physical_used.as_ref());
Ok(DecommissionPoolCapacityInfo {
pool_index: idx,
@@ -12442,7 +12620,20 @@ impl ECStore {
let mut capacity_infos = Vec::with_capacity(self.pools.len());
for idx in 0..self.pools.len() {
capacity_infos.push(self.get_decommission_pool_capacity_info(idx).await?);
capacity_infos.push(self.get_decommission_pool_capacity_info(idx, false).await?);
}
Ok(capacity_infos)
}
async fn get_decommission_all_pool_capacity_infos_with_owned_usage(&self) -> Result<Vec<DecommissionPoolCapacityInfo>> {
#[cfg(any(test, feature = "test-util"))]
if let Some(capacity_infos) = take_decommission_capacity_info_override_for_test(self.id) {
return Ok(capacity_infos);
}
let mut capacity_infos = Vec::with_capacity(self.pools.len());
for idx in 0..self.pools.len() {
capacity_infos.push(self.get_decommission_pool_capacity_info(idx, true).await?);
}
Ok(capacity_infos)
}
@@ -12544,15 +12735,37 @@ impl ECStore {
// different target so one mutation never holds two target gates.
discard_decommission_capacity_target_permit_except(self.id, owner, None);
}
reservation
if let Some((pool_index, _)) = reservation
.targets
.iter()
.filter_map(candidate)
.max_by_key(|(_, remaining)| *remaining)
.map(|(pool_index, _)| pool_index)
{
return Ok(pool_index);
}
if reservation.source_data_equivalent_bytes > 0 {
return Err(decommission_capacity_blocked_error(format!(
"source pool {} has no target allocation for {expected_data_bytes} data bytes",
owner.source_pool_index
)));
}
let capacities = self.get_decommission_all_pool_capacity_infos().await?;
let active_sources = active_decommission_source_indices(&pool_meta);
capacities
.into_iter()
.filter(|capacity| {
capacity.pool_index != owner.source_pool_index
&& !active_sources.contains(&capacity.pool_index)
&& pool_meta
.pools
.get(capacity.pool_index)
.is_some_and(is_decommission_start_active_pool)
})
.max_by_key(|capacity| capacity.physical_free)
.map(|capacity| capacity.pool_index)
.ok_or_else(|| {
decommission_capacity_blocked_error(format!(
"source pool {} has no target allocation for {expected_data_bytes} data bytes",
"source pool {} has no target pool for {expected_data_bytes} data bytes",
owner.source_pool_index
))
})
@@ -13048,7 +13261,7 @@ impl ECStore {
let capacity_infos = if self.pools.is_empty() {
Vec::new()
} else {
self.get_decommission_all_pool_capacity_infos().await?
self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?
};
let target_fence_proof = crate::services::notification_sys::acquire_decommission_target_fence_fleet_proof();
let mut pool_meta = self.pool_meta.write().await;
@@ -13749,6 +13962,20 @@ impl ECStore {
match self.acquire_decommission_capacity_target_guard(target_pool_index).await {
Ok(guard) => break guard,
Err(err) if is_decommission_capacity_target_gate_busy(&err) => {
if decommission_capacity_target_gate_retry_exhausted(wait_attempt) {
warn!(
event = EVENT_DECOMMISSION_STATE,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
pool_index = idx,
target_pool_index,
attempts = wait_attempt.saturating_add(1),
state = "capacity_gate_retry_exhausted",
error = %err,
"Decommission target capacity gate remained busy; pausing for supervised retry"
);
return Err(err);
}
wait_attempt = wait_attempt.saturating_add(1);
}
Err(err) => return Err(err),
@@ -15355,7 +15582,7 @@ impl ECStore {
.is_some_and(|reservation| !reservation.active())
};
if needs_capacity_reservation {
let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?;
let capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?;
let mut pool_meta = self.pool_meta.write().await;
let version = pool_meta.version;
pool_meta.version = POOL_META_VERSION;
@@ -16219,7 +16446,7 @@ impl ECStore {
self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate()
.await?;
let all_capacity_infos = self.get_decommission_all_pool_capacity_infos().await?;
let all_capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?;
let target_fence_proof_available =
crate::services::notification_sys::acquire_decommission_target_fence_fleet_proof().is_some();
// Signal cancellation before waiting for the movement writer so active
@@ -22653,6 +22880,7 @@ fn decommission_physical_pool_capacity(
pool_index: usize,
layout: DecommissionErasureLayout,
logical: PoolSpaceInfo,
owned_physical_used: Option<&HashMap<i32, usize>>,
) -> (usize, usize, usize) {
let width = layout.width();
let mut sets: HashMap<i32, Vec<&rustfs_madmin::Disk>> = HashMap::new();
@@ -22692,13 +22920,20 @@ fn decommission_physical_pool_capacity(
.map(|disk| disk.available_space as usize)
.min()
.unwrap_or_default();
let max_used = set_disks
.iter()
.map(|disk| (disk.used_space as usize).max((disk.total_space as usize).saturating_sub(disk.available_space as usize)))
.max()
.unwrap_or_default();
let physical_used_for_set = owned_physical_used
.and_then(|owned| owned.get(&set_disks[0].set_index).copied())
.unwrap_or_else(|| {
set_disks
.iter()
.map(|disk| {
(disk.used_space as usize).max((disk.total_space as usize).saturating_sub(disk.available_space as usize))
})
.max()
.unwrap_or_default()
.saturating_mul(width)
});
physical_total = physical_total.saturating_add(min_total.saturating_mul(width));
physical_used = physical_used.saturating_add(max_used.saturating_mul(width));
physical_used = physical_used.saturating_add(physical_used_for_set);
if set_disks.len() >= width {
physical_free = physical_free.saturating_add(min_free.saturating_mul(width));
}
@@ -22881,6 +23116,8 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
#[cfg(test)]
mod pools_tests {
use std::collections::HashMap;
use super::DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF;
use super::persist_v3_pool_meta_for_test;
use super::record_decommission_entry_error;
@@ -22914,10 +23151,11 @@ mod pools_tests {
ensure_decommission_start_rebalance_meta_allowed, ensure_decommission_start_target_capacity,
ensure_decommission_terminal_operation_supported, ensure_decommission_unresolved_verification_disk_count,
ensure_local_decommission_pool_leaders, ensure_pool_meta_write_fence, ensure_valid_decommission_pool_index, get_by_index,
guard_decommission_cancelers, has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
merge_decommission_durable_ilm_receipts, merge_pool_meta_updates_for_save, merge_pool_status_refresh,
missing_decommission_worker_prefix, next_decommission_capacity_generation, observe_decommission_terminal_reload_result,
grow_decommission_target_reservation, guard_decommission_cancelers, has_active_decommission_canceler,
is_decommission_active, is_decommission_cancel_requested, load_decommission_entry_versions,
local_decommission_queue_prefix, mark_decommission_bucket_done, merge_decommission_durable_ilm_receipts,
merge_pool_meta_updates_for_save, merge_pool_status_refresh, missing_decommission_worker_prefix,
next_decommission_capacity_generation, observe_decommission_terminal_reload_result,
parse_decommission_durable_ilm_receipt_path, pool_meta_has_active_decommission, publish_pool_meta_updates,
read_pool_meta_replica, reconcile_decommission_meta_buckets, reconcile_decommission_unresolved_entries_for_completion,
record_decommission_unresolved_entry, recover_decommission_capacity_reservations,
@@ -22943,8 +23181,9 @@ mod pools_tests {
with_decommission_entry_context,
};
use super::{
DecommissionCapacityAdmission, DecommissionCapacityOwner, DecommissionCapacityReleaseProof,
DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation, decommission_capacity_mutation_id,
DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS, DecommissionCapacityAdmission, DecommissionCapacityOwner,
DecommissionCapacityReleaseProof, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation,
decommission_capacity_mutation_id, decommission_capacity_target_gate_retry_exhausted,
ensure_decommission_target_owner_admission, ensure_exact_delete_capacity_namespace_fences,
ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error,
plan_exact_delete_capacity_reconciliations, record_decommission_target_consumption, release_decommission_target_inflight,
@@ -23051,6 +23290,17 @@ mod pools_tests {
identity: StdMutex<Option<(Vec<u8>, String)>>,
}
#[test]
fn decommission_target_gate_retry_is_bounded() {
assert!(!decommission_capacity_target_gate_retry_exhausted(0));
assert!(!decommission_capacity_target_gate_retry_exhausted(
DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS - 2
));
assert!(decommission_capacity_target_gate_retry_exhausted(
DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS - 1
));
}
#[tokio::test]
async fn pool_meta_recovery_reconciles_prepare_and_partial_commit_without_format_downgrade() {
for previous_version in [POOL_META_V1_VERSION, POOL_META_VERSION, POOL_META_GENERATION_VERSION] {
@@ -28472,6 +28722,44 @@ mod pools_tests {
}
}
#[test]
fn decommission_capacity_reservation_grows_after_stale_empty_snapshot() {
let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(1);
let layout = DecommissionErasureLayout { data: 3, parity: 1 };
let capacity_infos = vec![
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 0, 0),
DecommissionPoolCapacityInfo::for_test(1, layout, 64, 64, 0),
];
let mut meta = PoolMeta {
version: POOL_META_VERSION,
pools: vec![decommission_test_pool_status(0, None), decommission_test_pool_status(1, None)],
..Default::default()
};
meta.decommission(0, capacity_infos[0].space).unwrap();
reserve_decommission_start_target_capacity(
&mut meta,
&[0],
&capacity_infos,
uuid::Uuid::new_v4(),
1,
now,
DECOMMISSION_CAPACITY_MODEL_VERSION,
)
.expect("an empty scanner snapshot should still start decommission");
let grew = grow_decommission_target_reservation(&mut meta, &capacity_infos, 0, 1, 1, uuid::Uuid::new_v4(), now)
.expect("the reservation should grow against current target capacity");
assert!(grew);
let reservation = meta.pools[0]
.decommission
.as_ref()
.and_then(|info| info.capacity_reservation.as_ref())
.unwrap();
assert_eq!(reservation.source_data_equivalent_bytes, 2);
assert_eq!(reservation.predicted_physical_bytes, 3);
assert_eq!(reservation.targets.len(), 1);
assert_eq!(reservation.targets[0].reserved_physical_bytes, 6);
}
#[test]
fn decommission_capacity_model_reserves_versions_delete_markers_and_temporary_copies_from_physical_usage() {
let reservation = build_decommission_capacity_reservation(
@@ -29028,11 +29316,46 @@ mod pools_tests {
total: 0,
used: 0,
},
None,
);
assert_eq!(capacity, (400, 40, 360));
}
#[test]
fn decommission_physical_capacity_prefers_rustfs_owned_usage_over_filesystem_usage() {
let disks = [10_u64, 20, 30, 40]
.into_iter()
.enumerate()
.map(|(disk_index, available_space)| rustfs_madmin::Disk {
endpoint: format!("http://node-{disk_index}"),
drive_path: format!("/disk-{disk_index}"),
state: "ok".to_string(),
total_space: 100,
used_space: 100 - available_space,
available_space,
pool_index: 0,
set_index: 0,
disk_index: disk_index as i32,
..Default::default()
})
.collect::<Vec<_>>();
let owned = HashMap::from([(0, 12)]);
let capacity = decommission_physical_pool_capacity(
&disks,
0,
DecommissionErasureLayout { data: 2, parity: 2 },
PoolSpaceInfo {
free: 0,
total: 0,
used: 0,
},
Some(&owned),
);
assert_eq!(capacity, (400, 40, 12));
}
#[test]
fn decommission_terminal_transitions_release_capacity_reservations() {
let now = OffsetDateTime::UNIX_EPOCH + Duration::hours(1);
+17 -18
View File
@@ -244,13 +244,16 @@ impl StdError for ConditionalFileNotCommittedError {
}
}
fn classify_internode_missing_error(error: &InternodeHttpError) -> Option<DiskError> {
fn classify_internode_disk_error(error: &InternodeHttpError) -> Option<DiskError> {
if error.is_remote_file_not_found() {
return Some(DiskError::FileNotFound);
}
if error.is_remote_volume_not_found() {
return Some(DiskError::VolumeNotFound);
}
if error.is_remote_file_corrupt() {
return Some(DiskError::FileCorrupt);
}
None
}
@@ -526,13 +529,10 @@ fn io_error_chain_contains_kind(io_error: &std::io::Error, kind: std::io::ErrorK
impl From<std::io::Error> for DiskError {
fn from(e: std::io::Error) -> Self {
if let Some(error) = e.get_ref().and_then(|source| source.downcast_ref::<InternodeHttpError>()) {
if error.is_remote_file_not_found() {
return DiskError::FileNotFound;
}
if error.is_remote_volume_not_found() {
return DiskError::VolumeNotFound;
}
if let Some(error) = e.get_ref().and_then(|source| source.downcast_ref::<InternodeHttpError>())
&& let Some(classified) = classify_internode_disk_error(error)
{
return classified;
}
let e = match e.downcast::<TerminalReadError>() {
Ok(terminal_error) => {
@@ -541,7 +541,7 @@ impl From<std::io::Error> for DiskError {
&& let Some(internode_error) = io_error
.get_ref()
.and_then(|source| source.downcast_ref::<InternodeHttpError>())
&& let Some(classified) = classify_internode_missing_error(internode_error)
&& let Some(classified) = classify_internode_disk_error(internode_error)
{
return classified;
}
@@ -963,6 +963,7 @@ mod tests {
for (remote_error, expected) in [
(rustfs_rio::new_test_remote_file_not_found_http_io_error(), DiskError::FileNotFound),
(rustfs_rio::new_test_remote_volume_not_found_http_io_error(), DiskError::VolumeNotFound),
(rustfs_rio::new_test_remote_file_corrupt_http_io_error(), DiskError::FileCorrupt),
] {
let wrapped = terminal_read_error_to_io(DiskError::Io(remote_error));
assert_eq!(DiskError::from(wrapped), expected);
@@ -1533,25 +1534,23 @@ mod tests {
}
#[test]
fn test_internode_missing_errors_preserve_disk_error_types() {
fn test_internode_disk_errors_preserve_disk_error_types() {
let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error());
let volume_missing = DiskError::from(rustfs_rio::new_test_remote_volume_not_found_http_io_error());
let file_corrupt = DiskError::from(rustfs_rio::new_test_remote_file_corrupt_http_io_error());
let unmarked_server_error = DiskError::from(rustfs_rio::new_test_internode_http_io_error(
rustfs_rio::InternodeHttpErrorKind::HttpStatus(http::StatusCode::INTERNAL_SERVER_ERROR),
));
assert_eq!(file_missing, DiskError::FileNotFound);
assert_eq!(volume_missing, DiskError::VolumeNotFound);
assert_eq!(file_corrupt, DiskError::FileCorrupt);
assert!(matches!(unmarked_server_error, DiskError::Io(_)));
for missing in [file_missing, volume_missing] {
assert_eq!(missing.clone(), missing);
for error in [file_missing, volume_missing, file_corrupt] {
assert_eq!(error.clone(), error);
assert_eq!(
crate::disk::error_reduce::reduce_write_quorum_errs(
&[Some(missing.clone()), Some(missing.clone()), None],
&[],
2
),
Some(missing)
crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(error.clone()), Some(error.clone()), None], &[], 2),
Some(error)
);
}
}
+1 -1
View File
@@ -825,7 +825,7 @@ impl Disk {
dst_volume: &str,
dst_path: &str,
) -> Result<RenameDataResp> {
let Some(scope) = crate::store::bucket_heal_scope(dst_volume) else {
let Some(scope) = crate::store::bucket_heal_scope_for_object(dst_volume, dst_path) else {
return self
.rename_data_borrowed_with_fence(src_volume, src_path, fi, dst_volume, dst_path, None)
.await;
@@ -587,10 +587,8 @@ pub(crate) async fn verify_deep_parts(
return Err(DiskError::FileCorrupt);
}
let part_status = statuses.get_mut(&part_index).ok_or(DiskError::FileCorrupt)?;
// Deep verification reads the complete encoded shard. `shard_file_offset`
// is a range-read helper and can stop before the final encoded byte for
// a partial data block, which would let a one-byte tail truncation pass
// as healthy. Use the physical shard length for the integrity proof.
// Deep verification must consume every encoded stripe, including a
// partial final stripe, before certifying the shard's integrity.
let length = usize::try_from(erasure.shard_file_size(part.size as i64)).map_err(|_| DiskError::FileCorrupt)?;
let mut readers = Vec::with_capacity(disks.len());
for (index, disk) in disks.iter().enumerate() {
@@ -635,6 +633,9 @@ pub(crate) async fn verify_deep_parts(
part_status[index] = CHECK_PART_FILE_CORRUPT;
}
Err(error) => {
if !matches!(error, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::FileCorrupt) {
return Err(error);
}
readers.push(None);
part_status[index] = conv_part_err_to_int(&Some(error));
}
@@ -7,8 +7,9 @@ use super::meta::{
validate_start_rebalance_state,
};
use super::worker::{
resolve_rebalance_bucket_result, resolve_rebalance_meta_save_result, resolve_rebalance_save_task_result,
resolve_rebalance_terminal_error, send_rebalance_done_signal,
rebalance_max_attempts, resolve_rebalance_bucket_result, resolve_rebalance_meta_save_result,
resolve_rebalance_save_task_result, resolve_rebalance_terminal_error, retry_rebalance_metadata_access,
send_rebalance_done_signal,
};
use super::{
EVENT_REBALANCE_BUCKET, EVENT_REBALANCE_STATE, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REBALANCE,
@@ -449,13 +450,14 @@ impl ECStore {
};
if terminal_state_present {
if let Err(err) = store
.save_rebalance_stats_inner(
if let Err(err) = retry_rebalance_metadata_access(None, rebalance_max_attempts(), || {
store.save_rebalance_stats_inner(
pool_index,
RebalSaveOpt::Stats,
Some(save_rebalance_id.as_ref()),
)
.await
})
.await
{
let mut rebalance_meta = store.rebalance_meta.write().await;
*rebalance_meta = previous_meta;
@@ -475,9 +477,10 @@ impl ECStore {
}
if !terminal_state_saved
&& let Err(err) = store
.save_rebalance_stats_for_id(pool_index, RebalSaveOpt::Stats, save_rebalance_id.as_ref())
.await
&& let Err(err) = retry_rebalance_metadata_access(None, rebalance_max_attempts(), || {
store.save_rebalance_stats_for_id(pool_index, RebalSaveOpt::Stats, save_rebalance_id.as_ref())
})
.await
{
let wrapped = Error::other(format!("rebalance save_task stats save failed for pool {pool_index}: {err}"));
error!("{} err: {:?}", msg, wrapped);
@@ -109,7 +109,13 @@ pub(super) fn resolve_rebalance_save_task_result(
}
pub(super) fn resolve_rebalance_meta_save_result(result: Result<()>, stage: &str) -> Result<()> {
result.map_err(|err| Error::other(format!("rebalance meta save failed during {stage}: {err}")))
// Keep the source error reachable: the metadata retry policy classifies
// transient lock timeouts by inspecting the source chain, so collapsing the
// failure into a plain string here would make that retry a no-op.
result.map_err(|err| {
let rendered = format!("rebalance meta save failed during {stage}: {err}");
crate::data_movement::data_movement_context_error(rendered, err)
})
}
pub(super) fn rebalance_meta_lock_error(err: rustfs_lock::LockError, mode: &'static str) -> Error {
@@ -739,6 +745,29 @@ mod error_source_tests {
}
}
#[tokio::test]
async fn rebalance_metadata_retry_engages_for_wrapped_meta_save_lock_timeout() {
let mut attempts = 0;
let result = retry_rebalance_metadata_access(None, 3, || {
attempts += 1;
std::future::ready(resolve_rebalance_meta_save_result(
Err(rebalance_meta_lock_error(
rustfs_lock::LockError::timeout(".rustfs.sys/rebalance.bin@latest", Duration::from_secs(5)),
"write",
)),
"save_rebalance_stats for pool 0 opt Stats",
))
})
.await;
assert_eq!(attempts, 3, "a wrapped meta save lock timeout must stay retryable");
let err = result.expect_err("persistent lock contention must not become success");
assert!(matches!(
rebalance_error_source(&err),
Error::Lock(rustfs_lock::LockError::Timeout { .. })
));
}
#[tokio::test]
async fn rebalance_metadata_retry_cancels_a_pending_attempt() {
struct DropProbe(Arc<std::sync::atomic::AtomicUsize>);
+70
View File
@@ -6473,6 +6473,28 @@ async fn verify_inline_part_bitrot(meta: &FileInfo) -> disk::error::Result<()> {
.map_err(|_| DiskError::FileCorrupt)
}
fn validate_deep_scan_results(results: &[usize], expected_parts: usize) -> disk::error::Result<()> {
if results.len() != expected_parts {
return Err(DiskError::other(format!(
"incomplete deep scan result: expected {expected_parts} parts, received {}",
results.len()
)));
}
for (part, status) in results.iter().enumerate() {
match *status {
CHECK_PART_SUCCESS | CHECK_PART_FILE_NOT_FOUND | CHECK_PART_FILE_CORRUPT => {}
CHECK_PART_DISK_NOT_FOUND => return Err(DiskError::DiskNotFound),
crate::disk::CHECK_PART_VOLUME_NOT_FOUND => return Err(DiskError::VolumeNotFound),
_ => {
return Err(DiskError::other(format!(
"incomplete deep scan result: part {part} has unverified status {status}"
)));
}
}
}
Ok(())
}
/// disks_with_all_partsv2 is a corrected version based on Go implementation.
/// It sets partsMetadata and onlineDisks when xl.meta is inexistant/corrupted or outdated.
/// It also checks if the status of each part (corrupted, missing, ok) in each drive.
@@ -6693,9 +6715,13 @@ async fn disks_with_all_parts(
// it needs healing too.
match disk.verify_file(bucket, object, meta).await {
Ok(v) => {
validate_deep_scan_results(&v.results, latest_meta.parts.len())?;
verify_resp = v;
}
Err(err) => {
if !matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::FileCorrupt) {
return Err(err);
}
debug!(
event = EVENT_SET_DISK_HEAL,
component = LOG_COMPONENT_ECSTORE,
@@ -11109,6 +11135,50 @@ mod tests {
assert_eq!(conv_part_err_to_int(&Some(other_err)), CHECK_PART_UNKNOWN); // Other errors should return UNKNOWN, not SUCCESS
}
#[test]
fn deep_scan_results_accept_only_complete_verified_or_repairable_parts() {
for (results, expected_parts) in [
(vec![], 0),
(vec![CHECK_PART_SUCCESS], 1),
(vec![CHECK_PART_FILE_NOT_FOUND], 1),
(vec![CHECK_PART_FILE_CORRUPT], 1),
(vec![CHECK_PART_SUCCESS, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_FILE_CORRUPT], 3),
] {
validate_deep_scan_results(&results, expected_parts)
.expect("complete results must allow healthy or repairable parts");
}
}
#[test]
fn deep_scan_results_reject_unknown_and_incomplete_observations() {
for (results, expected_parts) in [
(vec![CHECK_PART_UNKNOWN], 1),
(vec![usize::MAX], 1),
(vec![CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN], 2),
(vec![], 1),
(vec![CHECK_PART_SUCCESS], 2),
(vec![CHECK_PART_SUCCESS, CHECK_PART_SUCCESS], 1),
(vec![CHECK_PART_SUCCESS], 0),
] {
let error =
validate_deep_scan_results(&results, expected_parts).expect_err("unverified parts must not become healthy");
assert!(matches!(error, DiskError::Io(_)), "an incomplete observation is not proven corruption");
assert!(error.to_string().contains("incomplete deep scan result"));
}
}
#[test]
fn deep_scan_results_preserve_disk_and_volume_failures() {
for (status, expected_error) in [
(CHECK_PART_DISK_NOT_FOUND, DiskError::DiskNotFound),
(CHECK_PART_VOLUME_NOT_FOUND, DiskError::VolumeNotFound),
] {
let error = validate_deep_scan_results(&[CHECK_PART_SUCCESS, status], 2)
.expect_err("an unavailable part cannot certify a healthy disk");
assert_eq!(error, expected_error);
}
}
#[test]
fn test_has_part_err() {
// Test checking for part errors
+445 -55
View File
@@ -854,7 +854,7 @@ impl SetDisks {
);
let disks = self.get_disks_internal().await;
let bucket_heal_scope = crate::store::bucket_heal_scope(bucket);
let bucket_heal_scope = crate::store::bucket_heal_scope_for_object(bucket, object);
if let Some(scope) = &bucket_heal_scope {
scope.check()?;
}
@@ -4378,63 +4378,243 @@ mod heal_result_report_tests {
);
}
#[tokio::test]
async fn deep_heal_rebuilds_h2_near_tail_truncated_part() {
let (temp_dirs, disks, set) = hermetic_set_disks_for_pool_with_default_parity_isolated(16, 0, 4).await;
let bucket = "deep-heal-h2-near-tail-truncation";
let object = "object.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("bucket volume should be created");
#[derive(Debug)]
struct FailedIntegrityReadTransport {
error: DiskError,
reads: std::sync::atomic::AtomicUsize,
}
#[async_trait::async_trait]
impl crate::cluster::rpc::internode_data_transport::InternodeDataTransport for FailedIntegrityReadTransport {
async fn open_read(
&self,
_request: crate::cluster::rpc::internode_data_transport::ReadStreamRequest,
) -> crate::disk::error::Result<crate::disk::FileReader> {
self.reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let error = match &self.error {
DiskError::FileCorrupt => DiskError::from(rustfs_rio::new_test_remote_file_corrupt_http_io_error()),
DiskError::FileNotFound => DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error()),
DiskError::VolumeNotFound => DiskError::from(rustfs_rio::new_test_remote_volume_not_found_http_io_error()),
error => error.clone(),
};
Err(error)
}
let expected_payload = vec![0x6d; 5 * 1024 * 1024 + 123];
async fn open_write(
&self,
_request: crate::cluster::rpc::internode_data_transport::WriteStreamRequest,
) -> crate::disk::error::Result<crate::disk::FileWriter> {
panic!("an integrity scan must not write remote data");
}
async fn open_walk_dir(
&self,
_request: crate::cluster::rpc::internode_data_transport::WalkDirStreamRequest,
) -> crate::disk::error::Result<crate::disk::FileReader> {
panic!("an object integrity scan must not walk remote directories");
}
fn name(&self) -> &'static str {
"failed-integrity-read"
}
fn capabilities(&self) -> crate::cluster::rpc::internode_data_transport::InternodeDataTransportCapabilities {
crate::cluster::rpc::internode_data_transport::InternodeDataTransportCapabilities::tcp_http()
}
}
async fn assert_remote_integrity_read_failure(expected_error: DiskError) {
use crate::disk::{CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN};
use crate::object_api::ShardIntegrityWriteMode;
let (dirs, disks, set) = hermetic_set_disks_for_pool_with_default_parity_isolated(4, 0, 2).await;
let bucket = "remote-integrity-read-failure";
let object = "object.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("create fixture bucket");
}
set.put_object(
bucket,
object,
&mut PutObjReader::from_vec(expected_payload.clone()),
&mut PutObjReader::from_vec(vec![0x6d; 1024 * 1024 + 123]),
&ObjectOptions {
shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected),
no_lock: true,
..Default::default()
},
)
.await
.expect("source object should be written before shard truncation");
let source = disks[2]
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("source metadata should be readable");
let data_dir = source.data_dir.expect("non-inline source should have a data directory");
let truncated_part = temp_dirs[1]
.path()
.join(bucket)
.join(object)
.join(data_dir.to_string())
.join("part.1");
let original_len = tokio::fs::metadata(&truncated_part)
.await
.expect("target shard should exist")
.len();
assert!(original_len > 1, "test shard must be large enough to truncate");
let file = tokio::fs::OpenOptions::new()
.write(true)
.open(&truncated_part)
.await
.expect("target shard should be writable");
file.set_len(original_len - 1)
.await
.expect("target shard should be truncated");
.expect("write protected external fixture");
let mut reader = set
.get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
.await
.expect("GET should remain readable after a one-byte shard truncation");
let mut read_back = Vec::new();
tokio::io::copy(&mut reader, &mut read_back)
.await
.expect("GET should reconstruct the truncated shard through EC");
assert_eq!(read_back, expected_payload, "EC GET must preserve the object bytes");
let mut files = Vec::new();
let mut snapshots = Vec::new();
for (dir, disk) in dirs.iter().zip(&disks) {
let file = disk
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("read fixture metadata");
assert!(file.data.is_none(), "the remote test must open an external shard");
let object_dir = dir.path().join(bucket).join(object);
let part_path = object_dir
.join(file.data_dir.expect("external shard directory").to_string())
.join("part.1");
for path in [part_path, object_dir.join("xl.meta")] {
let bytes = tokio::fs::read(&path).await.expect("snapshot fixture before scan");
snapshots.push((path, bytes));
}
files.push(file);
}
let transport = Arc::new(FailedIntegrityReadTransport {
error: expected_error.clone(),
reads: std::sync::atomic::AtomicUsize::new(0),
});
let endpoint = Endpoint {
url: url::Url::parse("http://remote-integrity.invalid:9000/data/disk0").expect("fixture endpoint"),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
};
let remote = crate::cluster::rpc::RemoteDisk::new(
&endpoint,
&DiskOption {
cleanup: false,
health_check: false,
},
transport.clone(),
)
.await
.expect("construct remote disk with failing data transport");
let mut verification_disks: Vec<_> = disks.iter().cloned().map(Some).collect();
verification_disks[0] = Some(Arc::new(crate::disk::Disk::Remote(Box::new(remote))));
let mut statuses = [(0, vec![CHECK_PART_UNKNOWN; disks.len()])].into_iter().collect();
let result = crate::io_support::shard_integrity::verify_deep_parts(
&files,
&verification_disks,
&files[0],
bucket,
object,
&mut statuses,
)
.await;
assert!(transport.reads.load(std::sync::atomic::Ordering::Relaxed) > 0);
if matches!(expected_error, DiskError::FileCorrupt | DiskError::FileNotFound) {
result.expect("typed remote payload damage must identify a repairable shard");
let expected_status = if expected_error == DiskError::FileCorrupt {
CHECK_PART_FILE_CORRUPT
} else {
CHECK_PART_FILE_NOT_FOUND
};
assert_eq!(statuses[&0][0], expected_status);
assert!(statuses[&0][1..].iter().all(|status| *status == CHECK_PART_SUCCESS));
let (heal, metadata, reason) = super::should_heal_object_on_disk(&None, &[statuses[&0][0]], &files[0], &files[0]);
assert!(heal, "remote payload damage must schedule repair: {expected_error:?}");
assert!(!metadata);
let expected_reason = if expected_error == DiskError::FileCorrupt {
DiskError::FileCorrupt
} else {
DiskError::PartMissingOrCorrupt
};
assert_eq!(reason, Some(expected_reason));
} else {
let error = result.expect_err("unavailable payload storage must abort health verification");
if expected_error.is_internode_http_status(500) {
assert!(error.is_internode_http_status(500), "preserve the original transport error: {error:?}");
} else {
assert_eq!(error, expected_error, "preserve the original storage failure");
}
assert_eq!(statuses[&0][0], CHECK_PART_UNKNOWN, "failed verification cannot certify the shard");
}
for (path, before) in snapshots {
assert_eq!(tokio::fs::read(path).await.expect("read fixture after scan"), before);
}
}
let (result, error) = set
#[tokio::test]
async fn deep_scan_remote_payload_damage_schedules_repair() {
for error in [DiskError::FileCorrupt, DiskError::FileNotFound] {
assert_remote_integrity_read_failure(error).await;
}
}
#[tokio::test]
async fn deep_scan_remote_storage_failure_cannot_certify_healthy() {
for error in [
DiskError::from(rustfs_rio::new_test_internode_http_io_error(
rustfs_rio::InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::INTERNAL_SERVER_ERROR),
)),
DiskError::DiskNotFound,
DiskError::VolumeNotFound,
DiskError::FileAccessDenied,
] {
assert_remote_integrity_read_failure(error).await;
}
}
#[cfg(unix)]
#[tokio::test]
async fn deep_heal_legacy_verification_error_cannot_certify_healthy() {
use crate::object_api::ShardIntegrityWriteMode;
let (dirs, disks, set) = hermetic_set_disks_isolated(4).await;
let bucket = "legacy-deep-scan-verification-error";
let object = "object.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("create legacy fixture bucket");
}
set.put_object(
bucket,
object,
&mut PutObjReader::from_vec(vec![0x6d; 1024 * 1024 + 123]),
&ObjectOptions {
shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Legacy),
no_lock: true,
..Default::default()
},
)
.await
.expect("write external legacy fixture");
let mut files = Vec::new();
let mut part_paths = Vec::new();
let mut preserved = Vec::new();
for (index, (dir, disk)) in dirs.iter().zip(&disks).enumerate() {
let file = disk
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("read legacy fixture metadata");
assert!(file.data.is_none(), "legacy verification must open the physical shard");
assert!(file.parts.iter().all(|part| part.integrity.is_none()));
let object_dir = dir.path().join(bucket).join(object);
let part = object_dir
.join(file.data_dir.expect("external legacy shard directory").to_string())
.join("part.1");
let metadata = object_dir.join("xl.meta");
let bytes = tokio::fs::read(&metadata).await.expect("snapshot legacy metadata");
preserved.push((metadata, bytes));
if index != 0 {
let bytes = tokio::fs::read(&part).await.expect("snapshot healthy legacy shard");
preserved.push((part.clone(), bytes));
}
part_paths.push(part);
files.push(file);
}
let broken_part = &part_paths[0];
tokio::fs::remove_file(broken_part)
.await
.expect("remove fixture shard before symlink fault");
std::os::unix::fs::symlink("part.1", broken_part).expect("create self-referential shard symlink");
let open_error = tokio::fs::File::open(broken_part)
.await
.expect_err("self-referential symlink must fail to open");
assert_eq!(open_error.raw_os_error(), Some(libc::ELOOP));
let verification_error = disks[0]
.verify_file(bucket, object, &files[0])
.await
.expect_err("legacy disk verifier must reject the symlink path");
assert_eq!(verification_error, DiskError::InvalidPath);
let error = set
.heal_object(
bucket,
object,
@@ -4446,19 +4626,229 @@ mod heal_result_report_tests {
},
)
.await
.expect("deep heal should finish after a one-byte H2 tail truncation");
assert!(error.is_none(), "deep heal should recover the H2 truncated shard: {error:?}");
assert_eq!(result.drives_healed(), Some(1));
assert_eq!(result.before.drives[1].state, DriveState::Corrupt.to_string());
.expect_err("an unverified legacy part must fail the complete heal operation");
assert_eq!(error, DiskError::InvalidPath, "complete heal must preserve the verifier's path error");
assert_eq!(
tokio::fs::metadata(&truncated_part)
tokio::fs::read_link(broken_part)
.await
.expect("repaired H2 shard should exist")
.len(),
original_len,
"deep heal must restore the complete H2 shard length"
.expect("failed heal must preserve the symlink fault"),
std::path::Path::new("part.1")
);
for (path, before) in preserved {
assert_eq!(tokio::fs::read(path).await.expect("read unchanged legacy fixture"), before);
}
}
async fn assert_h2_truncated_shards(
mode: crate::object_api::ShardIntegrityWriteMode,
coding_indexes: &[usize],
retained_len: fn(usize, usize) -> usize,
) {
use crate::object_api::ShardIntegrityWriteMode;
let (temp_dirs, disks, set) = hermetic_set_disks_for_pool_with_default_parity_isolated(16, 0, 4).await;
let bucket = "deep-heal-h2-truncation";
let object = "object.bin";
for disk in &disks {
disk.make_volume(bucket).await.expect("create bucket volume");
}
let expected_payload: Vec<_> = (0..5 * 1024 * 1024 + 123)
.map(|index| u8::try_from((index / 4096 + index) % 251).expect("payload byte fits"))
.collect();
set.put_object(
bucket,
object,
&mut PutObjReader::from_vec(expected_payload.clone()),
&ObjectOptions {
shard_integrity_write_mode: Some(mode),
no_lock: true,
..Default::default()
},
)
.await
.expect("commit all source shards");
let mut files = Vec::new();
let mut paths = Vec::new();
let mut original_shards = Vec::new();
let mut original_metadata = Vec::new();
let mut damaged = Vec::new();
for (slot, disk) in disks.iter().enumerate() {
let source = disk
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("read source metadata");
assert_eq!((source.erasure.data_blocks, source.erasure.parity_blocks), (12, 4));
assert_eq!(source.parts[0].integrity.is_some(), mode == ShardIntegrityWriteMode::Protected);
let directory = temp_dirs[slot].path().join(bucket).join(object);
let path = directory
.join(source.data_dir.expect("external shard data directory").to_string())
.join("part.1");
let original = tokio::fs::read(&path).await.expect("read original shard");
original_metadata.push(tokio::fs::read(directory.join("xl.meta")).await.expect("read xl.meta"));
if coding_indexes.contains(&source.erasure.index) {
let erasure = crate::erasure::coding::Erasure::try_new_with_options(
12,
4,
source.erasure.block_size,
source.uses_legacy_checksum,
)
.expect("valid erasure layout");
let stripe_size = erasure.shard_size() + source.erasure.get_checksum_info(1).algorithm.size();
let keep = retained_len(original.len(), stripe_size);
assert!(keep < original.len(), "fixture must truncate the shard");
tokio::fs::OpenOptions::new()
.write(true)
.open(&path)
.await
.expect("open target shard")
.set_len(u64::try_from(keep).expect("shard length fits"))
.await
.expect("truncate target shard");
damaged.push((slot, keep));
}
paths.push(path);
original_shards.push(original);
files.push(source);
}
assert_eq!(damaged.len(), coding_indexes.len());
if mode == ShardIntegrityWriteMode::Protected {
let mut statuses = [(0, vec![crate::disk::CHECK_PART_UNKNOWN; disks.len()])]
.into_iter()
.collect();
let online_disks: Vec<_> = disks.iter().cloned().map(Some).collect();
crate::io_support::shard_integrity::verify_deep_parts(
&files,
&online_disks,
&files[0],
bucket,
object,
&mut statuses,
)
.await
.expect("verify protected shards directly");
for (slot, status) in statuses[&0].iter().enumerate() {
let expected = if damaged.iter().any(|(index, _)| *index == slot) {
crate::disk::CHECK_PART_FILE_CORRUPT
} else {
crate::disk::CHECK_PART_SUCCESS
};
assert_eq!(*status, expected, "protected shard {slot}");
}
}
let opts = HealOpts {
no_lock: true,
scan_mode: HealScanMode::Deep,
..Default::default()
};
let (dry_run, error) = set
.heal_object(bucket, object, "", &HealOpts { dry_run: true, ..opts })
.await
.expect("dry-run should report corruption");
assert!(error.is_none(), "dry-run error: {error:?}");
assert_eq!(dry_run.drives_healed(), Some(0));
for (slot, keep) in &damaged {
assert_eq!(dry_run.after.drives[*slot].state, DriveState::Corrupt.to_string());
assert_eq!(
tokio::fs::read(&paths[*slot]).await.expect("dry-run preserved shard"),
original_shards[*slot][..*keep]
);
}
let recoverable = damaged.len() <= 4;
if recoverable {
let mut reader = set
.get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
.await
.expect("degraded GET should remain readable");
let mut read_back = Vec::new();
tokio::io::copy(&mut reader, &mut read_back)
.await
.expect("read complete degraded GET");
assert_eq!(read_back, expected_payload);
}
let (result, error) = set
.heal_object(bucket, object, "", &opts)
.await
.expect("heal should report outcome");
if recoverable {
assert!(error.is_none(), "heal error: {error:?}");
assert_eq!(result.drives_healed(), Some(damaged.len()));
for (slot, _) in &damaged {
assert_eq!(result.before.drives[*slot].state, DriveState::Corrupt.to_string());
}
} else {
assert_eq!(error, Some(DiskError::ErasureReadQuorum));
assert_eq!(result.drives_healed(), Some(0));
}
for (slot, path) in paths.iter().enumerate() {
let expected = if !recoverable && let Some((_, keep)) = damaged.iter().find(|(index, _)| *index == slot) {
&original_shards[slot][..*keep]
} else {
original_shards[slot].as_slice()
};
assert_eq!(
tokio::fs::read(path).await.expect("read shard after heal"),
expected,
"physical shard {slot}"
);
if !recoverable {
let metadata_path = temp_dirs[slot].path().join(bucket).join(object).join("xl.meta");
assert_eq!(tokio::fs::read(metadata_path).await.expect("preserved metadata"), original_metadata[slot]);
}
}
if recoverable {
let mut reader = set
.get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
.await
.expect("GET after repair");
let mut read_back = Vec::new();
tokio::io::copy(&mut reader, &mut read_back)
.await
.expect("read repaired object");
assert_eq!(read_back, expected_payload);
let (healthy, error) = set.heal_object(bucket, object, "", &opts).await.expect("second deep scan");
assert!(error.is_none());
assert_eq!(healthy.drives_healed(), Some(0), "repair must converge");
assert_eq!(healthy.integrity_verified, mode == ShardIntegrityWriteMode::Protected);
}
}
#[tokio::test]
async fn deep_heal_rebuilds_h2_near_tail_truncated_part() {
use crate::object_api::ShardIntegrityWriteMode::{Legacy, Protected};
for mode in [Legacy, Protected] {
for coding_index in [1, 13] {
assert_h2_truncated_shards(mode, &[coding_index], |length, _| length - 1).await;
}
}
}
#[tokio::test]
async fn deep_heal_rebuilds_h2_half_truncated_part() {
use crate::object_api::ShardIntegrityWriteMode::{Legacy, Protected};
for mode in [Legacy, Protected] {
for coding_index in [1, 13] {
assert_h2_truncated_shards(mode, &[coding_index], |length, _| length / 2).await;
}
}
}
#[tokio::test]
async fn deep_heal_rebuilds_h2_stripe_boundary_truncated_parts_at_read_quorum() {
use crate::object_api::ShardIntegrityWriteMode::{Legacy, Protected};
for mode in [Legacy, Protected] {
assert_h2_truncated_shards(mode, &[1, 2, 13, 16], |_, stripe| stripe).await;
}
}
#[tokio::test]
async fn deep_heal_rejects_h2_truncated_parts_below_read_quorum() {
use crate::object_api::ShardIntegrityWriteMode::{Legacy, Protected};
for mode in [Legacy, Protected] {
assert_h2_truncated_shards(mode, &[1, 2, 3, 13, 16], |length, _| length / 2).await;
}
}
#[tokio::test]
+309
View File
@@ -58,6 +58,26 @@ pub(crate) fn bucket_heal_scope(bucket: &str) -> Option<Arc<BucketHealScope>> {
.flatten()
}
/// Resolve only the two canonical bucket records; other internal paths carry no bucket authority.
pub(super) fn bucket_metadata_owner(object: &str) -> Option<&str> {
use crate::bucket::metadata::{BUCKET_INCARNATION_FILE, BUCKET_METADATA_FILE};
let (prefix, rest) = object.split_once('/')?;
let (bucket, file) = rest.split_once('/')?;
(prefix == disk::BUCKET_META_PREFIX
&& matches!(file, BUCKET_METADATA_FILE | BUCKET_INCARNATION_FILE)
&& check_valid_bucket_name_strict(bucket).is_ok())
.then_some(bucket)
}
pub(crate) fn bucket_heal_scope_for_object(bucket: &str, object: &str) -> Option<Arc<BucketHealScope>> {
let owner = if bucket == RUSTFS_META_BUCKET {
bucket_metadata_owner(object)?
} else {
bucket
};
bucket_heal_scope(owner)
}
/// Storage-owned proof for the exact version and every selected erasure location.
/// This is an in-process result, never reconstructed from admin drive telemetry.
#[derive(Debug)]
@@ -422,6 +442,97 @@ impl ECStore {
Ok(Arc::ptr_eq(&target_set, &pool.get_disks_by_key(POOL_META_NAME)))
}
/// Restore the bucket configuration and incarnation before retiring a replacement marker.
pub async fn heal_replacement_bucket_metadata(
self: &Arc<Self>,
bucket: &str,
opts: &HealOpts,
targets: &[String],
) -> Result<()> {
use crate::bucket::metadata::{BUCKET_INCARNATION_FILE, BUCKET_METADATA_FILE};
use crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in;
let (Some(pool_index), Some(set_index)) = (opts.pool, opts.set) else {
return Err(Error::PreconditionFailed);
};
if targets.is_empty() || opts.dry_run || opts.no_lock || check_valid_bucket_name_strict(bucket).is_err() {
return Err(Error::PreconditionFailed);
}
let pool = self
.pools
.get(pool_index)
.ok_or_else(|| invalid_heal_pool_index(pool_index, self.pools.len()))?;
let target_set = pool.get_disks_for_heal_object(BUCKET_METADATA_FILE, opts)?;
if targets.iter().any(|target| {
target_set
.set_endpoints
.iter()
.filter(|endpoint| endpoint.to_string() == *target)
.count()
!= 1
}) {
return Err(Error::PreconditionFailed);
}
let objects: Vec<_> = [BUCKET_METADATA_FILE, BUCKET_INCARNATION_FILE]
.into_iter()
.map(|file| format!("{}/{bucket}/{file}", disk::BUCKET_META_PREFIX))
.filter(|object| Arc::ptr_eq(&target_set, &pool.get_disks_by_key(object)))
.collect();
if objects.is_empty() {
return Ok(());
}
// Read the persisted identity without lazy migration: a replacement
// must never turn lost configuration into newly fabricated defaults.
let incarnation = self.bucket_incarnation_id_from_disk(bucket).await?;
let targets = targets.to_vec();
self.run_bucket_heal_at_incarnation(bucket, incarnation, opts, move |store, bucket, opts| async move {
// Match config writers: lifecycle, metadata transaction, object locks.
// Keep the transaction stable through target readback, including when
// the caller is cancelled and the storage owner finishes its write.
let transaction = acquire_bucket_metadata_transaction_read_lock_in(&store.ctx, &bucket).await?;
for object in objects {
let scope = bucket_heal_scope(&bucket).ok_or(Error::PreconditionFailed)?;
scope.check()?;
if transaction.is_lock_lost() {
return Err(Error::PreconditionFailed);
}
// Config objects occupy one pool, unlike pool.bin. Require a
// successful all-pool lookup before accepting another pool's ownership.
if !store.single_pool() {
let (owner, _) = store
.get_pool_info_for_delete_marker(RUSTFS_META_BUCKET, &object, &ObjectOptions::default())
.await?;
if owner.index != pool_index {
continue;
}
}
let metadata_opts = HealOpts { remove: false, ..opts };
let (result, error) = store.heal_object(RUSTFS_META_BUCKET, &object, "", &metadata_opts).await?;
if let Some(error) = error {
return Err(error);
}
for target in &targets {
let mut drives = result.after.drives.iter().filter(|drive| &drive.endpoint == target);
if !drives.next().is_some_and(|drive| drive.state == "ok") || drives.next().is_some() {
return Err(Error::PreconditionFailed);
}
}
if !store
.replacement_targets_have_version(RUSTFS_META_BUCKET, &object, "", pool_index, set_index, &targets)
.await?
{
return Err(Error::PreconditionFailed);
}
}
if transaction.is_lock_lost() {
return Err(Error::PreconditionFailed);
}
Ok(())
})
.await
}
/// Return every live erasure set selected by an object-heal scope.
pub async fn heal_erasure_set_scopes(&self, opts: &HealOpts) -> Result<Vec<(usize, usize)>> {
let pools = self.get_pools_for_heal_object(opts)?;
@@ -1345,6 +1456,16 @@ mod tests {
.await
.is_err()
);
for file in [".metadata.bin", ".bucket-incarnation"] {
let metadata = format!("buckets/{bucket}/{file}");
assert!(
store
.rename_local_data_at_incarnation(&disk_ref, (&bucket, object), &fi, (RUSTFS_META_BUCKET, &metadata), old)
.await
.is_err(),
"stale remote heal must not replace the successor's {file}"
);
}
assert!(store.heal_bucket_at_incarnation(&bucket, old, &opts).await.is_err());
assert!(
store
@@ -1743,6 +1864,194 @@ mod tests {
(temp_dir, store, shutdown)
}
#[test]
fn bucket_metadata_heal_owner_rejects_path_aliases_and_unrelated_records() {
for file in [".metadata.bin", ".bucket-incarnation"] {
assert_eq!(bucket_metadata_owner(&format!("buckets/example/{file}")), Some("example"));
}
for path in [
"/buckets/example/.metadata.bin",
"buckets/../.metadata.bin",
"buckets/example/../.metadata.bin",
"buckets/example/.metadata.bin/extra",
"buckets/example//.metadata.bin",
"buckets/example/.metadata.bin.old",
"config/iam/example/.metadata.bin",
"buckets/example/usage.json",
] {
assert_eq!(bucket_metadata_owner(path), None, "unexpected bucket authority for {path}");
}
}
#[tokio::test]
async fn replacement_bucket_metadata_respects_pool_ownership_and_required_records() {
let (root, store, shutdown) = multi_pool_heal_store().await;
let bucket = "replacement-bucket-metadata";
for invalid_bucket in ["MixedCase", "../example", RUSTFS_META_BUCKET] {
assert!(matches!(
store
.heal_replacement_bucket_metadata(
invalid_bucket,
&HealOpts {
pool: Some(0),
set: Some(0),
..Default::default()
},
&[root.path().join("pool0-disk0").to_string_lossy().into_owned()],
)
.await,
Err(Error::PreconditionFailed)
));
}
store
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("create bucket and its persisted records");
let mut owned = 0;
for pool in 0..2 {
let target = root.path().join(format!("pool{pool}-disk0"));
let records: Vec<_> = [".metadata.bin", ".bucket-incarnation"]
.into_iter()
.map(|file| {
let path = target.join(RUSTFS_META_BUCKET).join("buckets").join(bucket).join(file);
let exists = path.join("xl.meta").exists();
if exists {
std::fs::remove_dir_all(&path).expect("remove replacement metadata shard");
owned += 1;
}
(path, exists)
})
.collect();
store
.heal_replacement_bucket_metadata(
bucket,
&HealOpts {
pool: Some(pool),
set: Some(0),
recreate: true,
..Default::default()
},
&[target.to_string_lossy().into_owned()],
)
.await
.expect("repair owned records and accept authoritative ownership in the other pool");
for (path, existed) in records {
assert_eq!(path.join("xl.meta").exists(), existed, "wrong placement for {}", path.display());
}
}
assert_eq!(owned, 2, "each required record must have exactly one owning pool");
let object = format!("buckets/{bucket}/.metadata.bin");
let owner = store
.get_pool_idx_existing_with_opts(RUSTFS_META_BUCKET, &object, &ObjectOptions::default())
.await
.expect("persisted metadata owner");
let disk = store.pools[owner].disk_set[0].disks.read().await[0]
.clone()
.expect("metadata target disk");
let fi = disk
.read_version(
"",
RUSTFS_META_BUCKET,
&object,
"",
&disk::ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("inline metadata to repair through target-side admission");
assert!(fi.data.is_some(), "fixture must contain an inline metadata shard");
let target_path = root.path().join(format!("pool{owner}-disk0"));
std::fs::remove_dir_all(target_path.join(RUSTFS_META_BUCKET).join(&object)).expect("remove remote target shard");
let incarnation = store
.bucket_incarnation_id_from_disk(bucket)
.await
.expect("current incarnation");
store
.rename_local_data_at_incarnation(
&disk.endpoint().to_string(),
(disk::RUSTFS_META_TMP_BUCKET, "metadata-heal-control"),
&fi,
(RUSTFS_META_BUCKET, &object),
incarnation,
)
.await
.expect("the target-side RPC admission must accept current bucket metadata repair");
assert!(target_path.join(RUSTFS_META_BUCKET).join(&object).join("xl.meta").exists());
let (configuration, _) = read_config_no_lock_preserve_empty_with_metadata(store.clone(), &object)
.await
.expect("read the current bucket configuration");
save_config(store.clone(), &object, configuration)
.await
.expect("publish a newer config revision in the same bucket incarnation");
let committed =
std::fs::read(target_path.join(RUSTFS_META_BUCKET).join(&object).join("xl.meta")).expect("new target configuration");
assert!(
store
.rename_local_data_at_incarnation(
&disk.endpoint().to_string(),
(disk::RUSTFS_META_TMP_BUCKET, "delayed-metadata-heal"),
&fi,
(RUSTFS_META_BUCKET, &object),
incarnation,
)
.await
.is_err(),
"a delayed repair must not overwrite a newer config in the same incarnation"
);
assert_eq!(
std::fs::read(target_path.join(RUSTFS_META_BUCKET).join(&object).join("xl.meta")).expect("retained configuration"),
committed,
"rejected repair must leave the newer configuration unchanged"
);
for pool in 0..2 {
for disk in 0..4 {
let path = root
.path()
.join(format!("pool{pool}-disk{disk}"))
.join(RUSTFS_META_BUCKET)
.join("buckets")
.join(bucket)
.join(".metadata.bin");
if path.exists() {
std::fs::remove_dir_all(path).expect("remove every copy of required bucket configuration");
}
}
}
metadata_sys::remove_bucket_metadata_in(&store.ctx, bucket)
.await
.expect("remove the cached configuration so repair cannot rely on it");
assert!(
store
.heal_replacement_bucket_metadata(
bucket,
&HealOpts {
pool: Some(0),
set: Some(0),
recreate: true,
..Default::default()
},
&[root.path().join("pool0-disk0").to_string_lossy().into_owned()],
)
.await
.is_err(),
"total metadata loss must not be accepted as ownership in another pool"
);
for pool in 0..2 {
assert!(
!root
.path()
.join(format!("pool{pool}-disk0"))
.join(RUSTFS_META_BUCKET)
.join(&object)
.exists(),
"missing required configuration must not be recreated with defaults"
);
}
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial]
async fn absence_proof_requires_every_selected_pool() {
+1 -1
View File
@@ -426,8 +426,8 @@ mod bucket_fence;
pub(crate) use bucket::await_bucket_namespace_operation;
pub use bucket_fence::BucketIncarnationFenceGuard;
mod heal;
pub(crate) use heal::bucket_heal_scope;
pub use heal::{HealObjectAbsenceProof, HealObjectStorageResult};
pub(crate) use heal::{bucket_heal_scope, bucket_heal_scope_for_object};
mod heal_walk;
pub use heal_walk::HealWalkVersion;
mod init;
+40 -2
View File
@@ -161,14 +161,52 @@ impl ECStore {
"incarnation-bound rename requires namespace locking and a non-nil identity",
));
}
let bucket = if destination.0 == RUSTFS_META_BUCKET {
super::heal::bucket_metadata_owner(destination.1).ok_or(DiskError::FileAccessDenied)?
} else {
destination.0
};
let guard = Arc::new(
self.acquire_bucket_incarnation_fence(destination.0, expected)
self.acquire_bucket_incarnation_fence(bucket, expected)
.await
.map_err(|error| DiskError::other(error.to_string()))?,
);
if guard.is_lock_lost() {
return Err(DiskError::other("bucket heal incarnation fence was lost"));
}
let transaction = if destination.0 == RUSTFS_META_BUCKET {
// The receiver must retain the config fence after an RPC timeout.
// A delayed request must also reject a newer config in the same
// bucket incarnation before publishing its older repair payload.
let transaction = crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_read_lock_in(&self.ctx, bucket)
.await
.map_err(|error| DiskError::other(error.to_string()))?;
let (current, _) = self
.get_pool_info_for_delete_marker(
destination.0,
destination.1,
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
.map_err(|error| DiskError::other(error.to_string()))?;
let current = current.object_info;
if transaction.is_lock_lost()
|| guard.is_lock_lost()
|| fi.mod_time.is_none()
|| current.mod_time != fi.mod_time
|| current.data_dir != fi.data_dir
|| current.size != fi.size
|| current.etag != fi.get_etag()
{
return Err(DiskError::FileAccessDenied);
}
Some(transaction)
} else {
None
};
rename_local_data_with_ctx(
&self.ctx,
disk_ref,
@@ -176,7 +214,7 @@ impl ECStore {
fi,
destination,
RenameDataGuards {
external_guard: Some(guard),
external_guard: Some(Arc::new((guard, transaction))),
..Default::default()
},
)
+41
View File
@@ -907,6 +907,21 @@ impl ErasureSetHealer {
}
if failed_objects == 0 && skipped_objects == 0 && failed_buckets == 0 {
let targets = if self.pool_metadata_target_endpoints.is_empty() {
self.target_endpoints.as_ref()
} else {
self.pool_metadata_target_endpoints.as_ref()
};
if self.replacement_task_id.is_some() || (!self.heal_opts.dry_run && self.heal_opts.recreate && !targets.is_empty()) {
self.verify_replacement_identity_fence("bucket metadata").await?;
// Recheck even resumed buckets: their user-object cursor does not
// prove that the replacement holds the internal bucket records.
for bucket in buckets {
self.storage
.heal_replacement_bucket_metadata(bucket, &self.heal_opts, targets)
.await?;
}
}
self.heal_replacement_pool_metadata(
set_disk_id,
&mut ErasureSetPassCounters {
@@ -2267,6 +2282,8 @@ mod resume_loop_tests {
list_include_lifecycle_object_info: Mutex<Vec<bool>>,
replacement_target_identity_sequences: Mutex<VecDeque<Vec<ReplacementTargetIdentity>>>,
pool_metadata_placement: Mutex<Option<ReplacementCommitEvidence>>,
bucket_metadata_calls: Mutex<Vec<String>>,
bucket_metadata_failure: AtomicBool,
fail_listing: AtomicBool,
fail_listing_buckets: Mutex<HashSet<String>>,
}
@@ -2462,6 +2479,13 @@ mod resume_loop_tests {
None => Ok(true),
}
}
async fn heal_replacement_bucket_metadata(&self, bucket: &str, _opts: &HealOpts, _targets: &[String]) -> Result<()> {
self.bucket_metadata_calls.lock().unwrap().push(bucket.to_owned());
if self.bucket_metadata_failure.load(Ordering::SeqCst) {
return Err(Error::Storage(EcstoreError::PreconditionFailed));
}
Ok(())
}
async fn replacement_targets_have_version(
&self,
_bucket: &str,
@@ -3058,6 +3082,18 @@ mod resume_loop_tests {
)
.with_replacement_targets(vec!["replacement-a".to_string()], Some(replacement_task_id.clone()));
env.storage.bucket_metadata_failure.store(true, Ordering::SeqCst);
assert!(healer.heal_erasure_set(&["b".to_string()], "pool_0_set_0").await.is_err());
let incomplete = ResumeManager::load_replacement_intent(env.healer.disk.clone(), &replacement_task_id)
.await
.expect("failed metadata repair must retain replacement intent")
.get_state()
.await;
assert!(!incomplete.completed);
assert_ne!(incomplete.replacement_phase, crate::heal::resume::ReplacementPhase::Verified);
assert!(env.storage.calls().is_empty(), "metadata failure must stop completion before pool.bin");
env.storage.bucket_metadata_failure.store(false, Ordering::SeqCst);
let error = healer
.heal_erasure_set(&["b".to_string()], "pool_0_set_0")
.await
@@ -3073,6 +3109,11 @@ mod resume_loop_tests {
assert_eq!(state.replacement_phase, crate::heal::resume::ReplacementPhase::Intent);
assert_eq!(state.retry_count, 1);
assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
assert_eq!(
*env.storage.bucket_metadata_calls.lock().unwrap(),
vec!["b", "b"],
"resuming a completed user scan must retry bucket metadata"
);
}
#[tokio::test]
+12
View File
@@ -591,6 +591,11 @@ pub trait HealStorageAPI: Send + Sync {
Err(Error::other("replacement pool metadata placement is unsupported"))
}
/// Heal and physically verify bucket configuration on its replacement targets.
async fn heal_replacement_bucket_metadata(&self, _bucket: &str, _opts: &HealOpts, _targets: &[String]) -> Result<()> {
Err(Error::Storage(StorageError::PreconditionFailed))
}
/// Read target-specific physical evidence for one replacement version.
///
/// This is only used by automatic replacement healing after the normal
@@ -1595,6 +1600,13 @@ impl HealStorageAPI for ECStoreHealStorage {
.map_err(Error::Storage)
}
async fn heal_replacement_bucket_metadata(&self, bucket: &str, opts: &HealOpts, targets: &[String]) -> Result<()> {
self.ecstore
.heal_replacement_bucket_metadata(bucket, opts, targets)
.await
.map_err(Error::Storage)
}
async fn replacement_targets_have_version(
&self,
bucket: &str,
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Regression for replacement healing requiring pool.bin on a non-owning set.
//! Replacement healing must restore internal records only on their owning sets.
#![recursion_limit = "256"]
use http::HeaderMap;
@@ -101,6 +101,31 @@ async fn replacement_pool_metadata_follows_real_two_set_placement() {
storage.replacement_pool_metadata_required(&opts).is_err(),
"invalid scope must fail closed"
);
assert!(
storage
.heal_replacement_bucket_metadata(bucket, &opts, &[paths[0].to_string_lossy().into_owned()])
.await
.is_err(),
"bucket metadata repair must reject an incomplete or invalid replacement scope"
);
}
for targets in [vec![], vec![paths[4].to_string_lossy().into_owned()]] {
assert!(
storage
.heal_replacement_bucket_metadata(
bucket,
&HealOpts {
pool: Some(0),
set: Some(0),
..Default::default()
},
&targets,
)
.await
.is_err(),
"empty or wrong-set targets must be rejected before repairing metadata"
);
}
let mut owning_set = None;
@@ -120,6 +145,17 @@ async fn replacement_pool_metadata_follows_real_two_set_placement() {
assert!(owning_set.replace(set).is_none(), "only one set owns pool.bin");
std::fs::remove_dir_all(&metadata_path).unwrap();
}
let bucket_records: Vec<_> = [".metadata.bin", ".bucket-incarnation"]
.into_iter()
.map(|file| {
let path = target_path.join(RUSTFS_META_BUCKET).join("buckets").join(bucket).join(file);
let existed = path.join("xl.meta").exists();
if existed {
std::fs::remove_dir_all(&path).expect("remove bucket metadata shard from replacement");
}
(path, existed)
})
.collect();
// Select a user key using the real placement algorithm, independently
// of the metadata-scope decision under test.
let key = (0..1000)
@@ -191,6 +227,14 @@ async fn replacement_pool_metadata_follows_real_two_set_placement() {
"completed repair must clear its healing marker"
);
assert_eq!(metadata_path.join("xl.meta").exists(), owns_metadata);
for (path, owned) in bucket_records {
assert_eq!(
path.join("xl.meta").exists(),
owned,
"replacement must restore each owned bucket record without creating a wrong-set copy: {}",
path.display()
);
}
for (relative, bytes) in original_parts {
assert_eq!(std::fs::read(object_path.join(relative)).unwrap(), bytes, "reconstructed shard differs");
}
+16 -1
View File
@@ -241,7 +241,9 @@ impl RuleValidate for LifecycleRule {
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS));
}
// Rule must have at least one action
let has_expiration = self.expiration.is_some();
let has_expiration = self.expiration.as_ref().is_some_and(|expiration| {
expiration.days.is_some() || expiration.date.is_some() || expiration.expired_object_delete_marker.is_some()
});
let has_transition = self.transitions.as_ref().is_some_and(|t| !t.is_empty());
// `NewerNoncurrentVersions` on its own is a MinIO extension, not an AWS
// form: it keeps the newest N noncurrent versions and expires the rest
@@ -4455,6 +4457,19 @@ mod tests {
.expect_err("a negative retention count on a transition must be rejected");
}
#[tokio::test]
async fn validate_rejects_empty_expiration_action() {
let mut rule = rule_with_filter(LifecycleRuleFilter::default());
rule.expiration = Some(LifecycleExpiration::default());
let err = config_with_rules(vec![rule])
.validate(&ObjectLockConfiguration::default())
.await
.expect_err("an empty Expiration object must not count as an action");
assert_eq!(err.to_string(), ERR_LIFECYCLE_RULE_MUST_HAVE_ACTION);
}
#[tokio::test]
async fn zero_newer_noncurrent_versions_means_no_count_constraint() {
// Zero carries no constraint, matching how the batch limit path has
+12
View File
@@ -89,6 +89,18 @@ fn rust_log_requests_verbose(rust_log: &str) -> bool {
})
}
/// Returns `true` when the effective logging configuration selects a verbose
/// (`debug`/`trace`) level.
///
/// Follows the same precedence as [`build_env_filter`]: `RUST_LOG` overrides
/// `logger_level` when it is set.
pub(super) fn effective_verbose_logging(logger_level: &str) -> bool {
match env::var("RUST_LOG").ok().filter(|value| !value.trim().is_empty()) {
Some(rust_log) => rust_log_requests_verbose(&rust_log),
None => is_verbose_level(logger_level),
}
}
fn should_suppress_noisy_crates(logger_level: &str, default_level: Option<&str>, rust_log: Option<&str>) -> bool {
if let Some(level) = default_level {
return !is_verbose_level(level);
+143 -13
View File
@@ -36,7 +36,7 @@ use crate::cleaner::LogCleaner;
use crate::cleaner::types::{CompressionAlgorithm, FileMatchMode};
use crate::config::OtelConfig;
use crate::global::{METRIC_LOG_CLEANER_RUN_FAILURES_TOTAL, METRIC_LOG_CLEANER_RUNS_TOTAL, set_observability_metric_enabled};
use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter};
use crate::telemetry::filter::{build_env_filter, effective_verbose_logging, pyroscope_log_filter};
use crate::telemetry::rolling::{RollingAppender, Rotation};
use metrics::counter;
use rustfs_config::observability::{
@@ -44,10 +44,11 @@ use rustfs_config::observability::{
DEFAULT_OBS_LOG_COMPRESSION_ALGORITHM, DEFAULT_OBS_LOG_DELETE_EMPTY_FILES, DEFAULT_OBS_LOG_DRY_RUN,
DEFAULT_OBS_LOG_GZIP_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_MATCH_MODE, DEFAULT_OBS_LOG_MAX_SINGLE_FILE_SIZE_BYTES,
DEFAULT_OBS_LOG_MAX_TOTAL_SIZE_BYTES, DEFAULT_OBS_LOG_MIN_FILE_AGE_SECONDS, DEFAULT_OBS_LOG_PARALLEL_COMPRESS,
DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL, DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP,
DEFAULT_OBS_LOG_ZSTD_WORKERS,
DEFAULT_OBS_LOG_PARALLEL_WORKERS, DEFAULT_OBS_LOG_SPAN_EVENTS, DEFAULT_OBS_LOG_ZSTD_COMPRESSION_LEVEL,
DEFAULT_OBS_LOG_ZSTD_FALLBACK_TO_GZIP, DEFAULT_OBS_LOG_ZSTD_WORKERS, ENV_OBS_LOG_SPAN_EVENTS,
};
use rustfs_config::{APP_NAME, DEFAULT_LOG_KEEP_FILES, DEFAULT_LOG_ROTATION_TIME};
use rustfs_utils::get_env_str;
use serde_json::Value as JsonValue;
use std::sync::Arc;
use std::{collections::BTreeMap, fmt};
@@ -78,6 +79,37 @@ pub(super) fn resolve_file_stdout_mirror(configured: Option<bool>, is_production
configured.unwrap_or(!is_production)
}
/// Resolves the span lifecycle events emitted by the JSON log layers.
///
/// Defaults to [`FmtSpan::NONE`]. A record per span lifecycle multiplies log
/// volume on object hot paths and, combined with the ancestor span list,
/// produced multi-megabyte lines that journald truncated (backlog#2642).
/// Operators opt back in with `RUSTFS_OBS_LOG_SPAN_EVENTS=close|full`.
pub(super) fn resolve_span_events() -> FmtSpan {
match get_env_str(ENV_OBS_LOG_SPAN_EVENTS, DEFAULT_OBS_LOG_SPAN_EVENTS)
.trim()
.to_ascii_lowercase()
.as_str()
{
"close" => FmtSpan::CLOSE,
"full" => FmtSpan::FULL,
"none" | "" => FmtSpan::NONE,
unknown => {
// An unrecognised value must not silently re-enable span events.
eprintln!("{STDERR_WARNING_PREFIX} obs: unknown {ENV_OBS_LOG_SPAN_EVENTS}={unknown:?}; falling back to none");
FmtSpan::NONE
}
}
}
/// Whether JSON log records embed the full ancestor span list.
///
/// `with_span_list(true)` re-serializes every ancestor span on each record, so
/// it is reserved for verbose levels where that context earns its volume.
pub(super) fn resolve_span_list(logger_level: &str) -> bool {
effective_verbose_logging(logger_level)
}
#[cfg(unix)]
pub(super) fn validate_stdout_sink(file_appender: &RollingAppender) -> Result<(), TelemetryError> {
use std::os::fd::AsFd;
@@ -222,7 +254,12 @@ where
request_id
}
pub(super) fn build_json_log_layer<S, W>(writer: W, enable_ansi: bool, span_events: FmtSpan) -> impl tracing_subscriber::Layer<S>
pub(super) fn build_json_log_layer<S, W>(
writer: W,
enable_ansi: bool,
span_events: FmtSpan,
span_list: bool,
) -> impl tracing_subscriber::Layer<S>
where
S: Subscriber + for<'span> LookupSpan<'span>,
W: for<'writer> tracing_subscriber::fmt::MakeWriter<'writer> + Send + Sync + 'static,
@@ -239,7 +276,7 @@ where
.json()
.flatten_event(true)
.with_current_span(true)
.with_span_list(true)
.with_span_list(span_list)
.with_span_events(span_events)
.map_event_format(RequestIdJsonFormat::new)
}
@@ -311,11 +348,12 @@ pub(super) fn init_local_logging(
fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: bool) -> Result<OtelGuard, TelemetryError> {
let env_filter = build_env_filter(logger_level, None);
let (nb, guard) = tracing_appender::non_blocking(std::io::stdout());
let span_events = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
let span_events = resolve_span_events();
let span_list = resolve_span_list(logger_level);
// Keep stdout formatting JSON-shaped even in local-only mode so operators
// can ship the same log schema to external collectors if needed.
let fmt_layer = build_json_log_layer(nb, std::io::stdout().is_terminal(), span_events);
let fmt_layer = build_json_log_layer(nb, std::io::stdout().is_terminal(), span_events, span_list);
tracing_subscriber::registry()
.with(env_filter)
@@ -413,11 +451,12 @@ fn init_file_logging_internal(
// ── 4. Build subscriber layers ────────────────────────────────────────────
let env_filter = build_env_filter(logger_level, None);
let span_events = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
let span_events = resolve_span_events();
let span_list = resolve_span_list(logger_level);
// File output stays machine-readable and free of ANSI sequences so the
// resulting files are safe to parse or ship to log processors.
let file_layer = build_json_log_layer(non_blocking, false, span_events.clone());
let file_layer = build_json_log_layer(non_blocking, false, span_events.clone(), span_list);
// Optional stdout mirror: enabled explicitly via `log_stdout_enabled`, or
// unconditionally in non-production environments so developers still see
@@ -425,7 +464,10 @@ fn init_file_logging_internal(
let (stdout_layer, stdout_guard) = if resolve_file_stdout_mirror(config.log_stdout_enabled, is_production) {
let (stdout_nb, stdout_guard) = tracing_appender::non_blocking(std::io::stdout());
let enable_color = std::io::stdout().is_terminal();
(Some(build_json_log_layer(stdout_nb, enable_color, span_events)), Some(stdout_guard))
(
Some(build_json_log_layer(stdout_nb, enable_color, span_events, span_list)),
Some(stdout_guard),
)
} else {
(None, None)
};
@@ -773,7 +815,7 @@ mod tests {
fn render_json_log(enable_ansi: bool) -> (String, Value) {
let writer = SharedWriter::default();
let layer = build_json_log_layer(writer.clone(), enable_ansi, FmtSpan::NONE);
let layer = build_json_log_layer(writer.clone(), enable_ansi, FmtSpan::NONE, false);
let subscriber = Registry::default().with(layer);
tracing::subscriber::with_default(subscriber, || {
@@ -793,9 +835,32 @@ mod tests {
(raw, parsed)
}
fn render_event_in_nested_spans(depth: usize, span_list: bool) -> String {
let writer = SharedWriter::default();
let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE, span_list);
let subscriber = Registry::default().with(layer);
tracing::subscriber::with_default(subscriber, || {
let spans = (0..depth)
.map(|level| tracing::info_span!("layer", depth = level, payload = "x".repeat(1024)))
.collect::<Vec<_>>();
let _guards = spans.iter().map(tracing::Span::enter).collect::<Vec<_>>();
info!(component = "obs_contract_test", "nested event");
});
String::from_utf8(
writer
.inner
.lock()
.expect("shared writer lock should not be poisoned")
.clone(),
)
.expect("log output should be valid UTF-8")
}
fn render_json_log_with_request_span() -> Value {
let writer = SharedWriter::default();
let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE);
let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE, false);
let subscriber = Registry::default().with(layer);
tracing::subscriber::with_default(subscriber, || {
@@ -818,7 +883,7 @@ mod tests {
fn render_json_log_with_recovery_monitor_child_span() -> Value {
let writer = SharedWriter::default();
let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE);
let layer = build_json_log_layer(writer.clone(), false, FmtSpan::NONE, true);
let subscriber = Registry::default().with(layer);
tracing::subscriber::with_default(subscriber, || {
@@ -1075,6 +1140,10 @@ mod tests {
assert_eq!(parsed["request-id"], Value::String("req-123".to_string()));
assert_eq!(parsed["message"], Value::String("inside request span".to_string()));
assert_eq!(parsed["span"]["request_id"], Value::String("req-123".to_string()));
assert!(
parsed.get("spans").is_none(),
"ancestor span list must stay out of records when span_list is disabled: {parsed}"
);
}
#[test]
@@ -1112,4 +1181,65 @@ mod tests {
assert_eq!(resolve_log_cleanup_file_pattern(&config, "rustfs.log"), "rustfs.log");
}
/// Regression for backlog#2642: records must not embed the ancestor span
/// list, so their size stays flat as the enclosing span chain deepens.
#[test]
fn record_size_does_not_grow_with_span_depth_when_span_list_is_disabled() {
let shallow = render_event_in_nested_spans(1, false);
let deep = render_event_in_nested_spans(32, false);
assert!(
deep.len() <= shallow.len() * 2,
"ancestor spans leaked into the record: shallow={} deep={}",
shallow.len(),
deep.len()
);
// The flag is what keeps it bounded: re-enabling the list must show the
// growth the default avoids.
let deep_with_list = render_event_in_nested_spans(32, true);
assert!(
deep_with_list.len() > deep.len(),
"enabling the span list must reintroduce per-ancestor growth"
);
}
#[test]
fn span_events_default_to_none_and_parse_overrides() {
temp_env::with_var(ENV_OBS_LOG_SPAN_EVENTS, None::<&str>, || {
assert_eq!(resolve_span_events(), FmtSpan::NONE);
});
temp_env::with_var(ENV_OBS_LOG_SPAN_EVENTS, Some("close"), || {
assert_eq!(resolve_span_events(), FmtSpan::CLOSE);
});
temp_env::with_var(ENV_OBS_LOG_SPAN_EVENTS, Some("FULL"), || {
assert_eq!(resolve_span_events(), FmtSpan::FULL);
});
temp_env::with_var(ENV_OBS_LOG_SPAN_EVENTS, Some("verbose"), || {
assert_eq!(
resolve_span_events(),
FmtSpan::NONE,
"an unrecognised value must not re-enable span events"
);
});
}
#[test]
fn span_list_is_reserved_for_verbose_levels() {
temp_env::with_var("RUST_LOG", None::<&str>, || {
assert!(!resolve_span_list("info"), "info records must not embed the ancestor span list");
assert!(!resolve_span_list("warn"));
assert!(!resolve_span_list("error"));
assert!(resolve_span_list("debug"));
assert!(resolve_span_list("trace"));
});
temp_env::with_var("RUST_LOG", Some("debug"), || {
assert!(
resolve_span_list("info"),
"RUST_LOG=debug overrides the base level and should keep span context"
);
});
}
}
+5 -4
View File
@@ -68,7 +68,7 @@ use std::{fs, io::IsTerminal, time::Duration};
use tracing::{info, warn};
use tracing_error::ErrorLayer;
use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
use tracing_subscriber::{fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
const GET_OBJECT_DURATION_HISTOGRAM_METRICS: &[&str] = &[
"rustfs_io_get_object_request_duration_seconds",
@@ -184,7 +184,8 @@ pub(super) fn init_observability_http(
// active, the OpenTelemetry bridge is the authoritative sink for
// `tracing` events unless local file logging is needed as a fallback.
}
let span_events = if is_production { FmtSpan::CLOSE } else { FmtSpan::FULL };
let span_events = crate::telemetry::local::resolve_span_events();
let span_list = crate::telemetry::local::resolve_span_list(logger_level);
// ── Case 2: File Logging
// If a log directory is configured and OTLP log export is unavailable, use
// the same rolling-file behavior as the local-only telemetry backend.
@@ -221,7 +222,7 @@ pub(super) fn init_observability_http(
crate::telemetry::local::validate_stdout_sink(&file_appender)?;
let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
let file_layer = build_json_log_layer(non_blocking, false, span_events.clone());
let file_layer = build_json_log_layer(non_blocking, false, span_events.clone(), span_list);
let cleanup_handle = spawn_cleanup_task(config, log_directory, log_filename, keep_files);
Ok((file_layer, guard, cleanup_handle, rotation_str))
})();
@@ -267,7 +268,7 @@ pub(super) fn init_observability_http(
if force_stdout_logging || crate::telemetry::local::resolve_file_stdout_mirror(config.log_stdout_enabled, is_production) {
let (stdout_nb, stdout_g) = tracing_appender::non_blocking(std::io::stdout());
stdout_guard = Some(stdout_g);
stdout_layer_opt = Some(build_json_log_layer(stdout_nb, std::io::stdout().is_terminal(), span_events));
stdout_layer_opt = Some(build_json_log_layer(stdout_nb, std::io::stdout().is_terminal(), span_events, span_list));
}
let local_file_fallback_enabled = file_layer_opt.is_some();
let stdout_mirror_enabled = stdout_guard.is_some();
+112
View File
@@ -57,6 +57,7 @@ const EXCESSIVE_EMPTY_CHUNKS_ERROR: &str = "HTTP body returned too many empty ch
pub const INTERNODE_DISK_ERROR_HEADER: &str = "x-rustfs-disk-error";
pub const INTERNODE_FILE_NOT_FOUND: &str = "file-not-found";
pub const INTERNODE_VOLUME_NOT_FOUND: &str = "volume-not-found";
pub const INTERNODE_FILE_CORRUPT: &str = "file-corrupt";
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum InternodeHttpErrorKind {
@@ -185,6 +186,7 @@ pub struct InternodeHttpError {
enum RemoteDiskErrorKind {
FileNotFound,
VolumeNotFound,
FileCorrupt,
}
impl std::fmt::Debug for InternodeHttpError {
@@ -215,6 +217,10 @@ impl InternodeHttpError {
self.remote_disk_error == Some(RemoteDiskErrorKind::VolumeNotFound)
}
pub fn is_remote_file_corrupt(&self) -> bool {
self.remote_disk_error == Some(RemoteDiskErrorKind::FileCorrupt)
}
fn new(kind: InternodeHttpErrorKind, context: InternodeHttpRequestContext) -> Self {
Self {
kind,
@@ -328,6 +334,20 @@ pub fn new_test_remote_volume_not_found_http_io_error() -> io::Error {
.into_io_error()
}
#[doc(hidden)]
pub fn new_test_remote_file_corrupt_http_io_error() -> io::Error {
InternodeHttpError::with_remote_disk_error(
InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::INTERNAL_SERVER_ERROR),
InternodeHttpRequestContext {
method: "GET".to_string(),
target: READ_FILE_STREAM_PATH.to_string(),
operation: Some(INTERNODE_OPERATION_READ_FILE_STREAM),
},
RemoteDiskErrorKind::FileCorrupt,
)
.into_io_error()
}
fn add_root_certificates_from_der(builder: reqwest::ClientBuilder, certs_der: &[Vec<u8>]) -> reqwest::ClientBuilder {
let mut b = builder;
for der in certs_der {
@@ -850,6 +870,7 @@ fn classify_http_response(
let remote_disk_error = match headers.get(INTERNODE_DISK_ERROR_HEADER).and_then(|value| value.to_str().ok()) {
Some(INTERNODE_FILE_NOT_FOUND) => Some(RemoteDiskErrorKind::FileNotFound),
Some(INTERNODE_VOLUME_NOT_FOUND) => Some(RemoteDiskErrorKind::VolumeNotFound),
Some(INTERNODE_FILE_CORRUPT) => Some(RemoteDiskErrorKind::FileCorrupt),
_ => None,
};
ClassifiedHttpResponse { kind, remote_disk_error }
@@ -2047,6 +2068,7 @@ mod tests {
assert_eq!(INTERNODE_DISK_ERROR_HEADER, "x-rustfs-disk-error");
assert_eq!(INTERNODE_FILE_NOT_FOUND, "file-not-found");
assert_eq!(INTERNODE_VOLUME_NOT_FOUND, "volume-not-found");
assert_eq!(INTERNODE_FILE_CORRUPT, "file-corrupt");
}
#[test]
@@ -2090,6 +2112,48 @@ mod tests {
assert!(wrong_operation.remote_disk_error.is_none());
}
#[test]
fn classify_http_response_scopes_corrupt_disk_error_to_read_failures() {
let mut headers = HeaderMap::new();
headers.insert(INTERNODE_DISK_ERROR_HEADER, INTERNODE_FILE_CORRUPT.parse().expect("valid error token"));
let classified = classify_http_response(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
&headers,
Some(INTERNODE_OPERATION_READ_FILE_STREAM),
);
assert_eq!(classified.remote_disk_error, Some(RemoteDiskErrorKind::FileCorrupt));
for status in [reqwest::StatusCode::NOT_FOUND, reqwest::StatusCode::SERVICE_UNAVAILABLE] {
assert!(
classify_http_response(status, &headers, Some(INTERNODE_OPERATION_READ_FILE_STREAM))
.remote_disk_error
.is_none()
);
}
for operation in [
None,
Some(INTERNODE_OPERATION_WALK_DIR),
Some(INTERNODE_OPERATION_PUT_FILE_STREAM),
] {
assert!(
classify_http_response(reqwest::StatusCode::INTERNAL_SERVER_ERROR, &headers, operation)
.remote_disk_error
.is_none()
);
}
headers.insert(INTERNODE_DISK_ERROR_HEADER, "unknown-disk-error".parse().expect("valid unknown token"));
for headers in [&headers, &HeaderMap::new()] {
assert!(
classify_http_response(
reqwest::StatusCode::INTERNAL_SERVER_ERROR,
headers,
Some(INTERNODE_OPERATION_READ_FILE_STREAM),
)
.remote_disk_error
.is_none()
);
}
}
#[derive(Clone, Default)]
struct TestState {
head_count: Arc<AtomicUsize>,
@@ -2158,6 +2222,16 @@ mod tests {
let addr = listener.local_addr().expect("listener local address should be available");
let app = Router::new()
.route("/stream", get(get_stream).head(reject_head).put(accept_put))
.route(
READ_FILE_STREAM_PATH,
get(|| async {
(
StatusCode::INTERNAL_SERVER_ERROR,
[(INTERNODE_DISK_ERROR_HEADER, INTERNODE_FILE_CORRUPT)],
"read file err file corrupt",
)
}),
)
.route(WALK_DIR_PATH, get(get_stream))
.route("/reject-put", get(get_stream).put(reject_put))
.route("/stall", get(get_stalling_stream))
@@ -2172,6 +2246,44 @@ mod tests {
Some((format!("http://{addr}/stream"), handle))
}
#[tokio::test]
async fn http_readers_preserve_remote_file_corruption() {
let (url, server) = start_test_server(TestState::default())
.await
.expect("corruption regression server must bind");
let url = format!("{}{READ_FILE_STREAM_PATH}", url.strip_suffix("/stream").expect("test server URL suffix"));
let byte_error = HttpReader::new(url.clone(), Method::GET, HeaderMap::new(), None)
.await
.err()
.expect("byte reader must reject corrupt shard response");
let chunk_error = HttpChunkReader::new_with_stall_timeout(url, Method::GET, HeaderMap::new(), None, None)
.await
.err()
.expect("chunk reader must reject corrupt shard response");
for error in [byte_error, chunk_error] {
let source = error
.get_ref()
.and_then(|source| source.downcast_ref::<InternodeHttpError>())
.expect("HTTP error must retain typed internode source");
assert!(source.is_remote_file_corrupt());
assert!(!source.is_remote_file_not_found());
assert!(!source.is_remote_volume_not_found());
assert_eq!(
source.kind(),
InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::INTERNAL_SERVER_ERROR)
);
let cloned = clone_internode_http_io_error(&error).expect("typed transport error must be cloneable");
assert!(
cloned
.get_ref()
.and_then(|source| source.downcast_ref::<InternodeHttpError>())
.expect("clone must retain typed internode source")
.is_remote_file_corrupt()
);
}
server.abort();
}
struct BlockedH2Server {
url: String,
shutdown: tokio::sync::oneshot::Sender<()>,
@@ -41,7 +41,7 @@ Every route asserted in `rustfs/src/admin/route_registration_test.rs` is registe
| Pools decommission / cancel / clear | 部分兼容 | `rustfs/src/admin/handlers/pools.rs` returns `NotImplemented` when endpoints are not initialized (single-pool or uninitialized clusters). |
| `/v3/top/drives`, `/v3/top/net` | 缺失 | Only `/v3/top/locks` is registered (`rustfs/src/admin/handlers/diagnostics.rs`). |
| Bucket / site replication per-object diff | 缺失 | `/v3/replicationmetrics` and site-replication status exist; no diff endpoint. |
| MRF (most-recent-failures) replication metrics breakdown | 缺失 | Only the generic `/v3/metrics` stream and replication metrics wire (`rustfs/src/admin/replication_metrics_wire.rs`). |
| MRF (most-recent-failures) replication metrics breakdown | 缺失 | Only the generic `/v3/realtime` stream and replication metrics wire (`rustfs/src/admin/replication_metrics_wire.rs`). |
Formerly-missing families that are now registered and therefore not exceptions: `/v3/healthinfo`, `/v3/obdinfo`, `/v3/force-unlock`, `/v3/top/locks`, `/v3/speedtest*`, `/v3/log`, `/v3/trace`, `/v3/profile`, `/v3/profiling/*`, `/v3/idp/{ldap|openid}/*`, `/v3/idp-config/*`.
@@ -11,7 +11,7 @@ Every stage is env-gated, so "before" and "after" run the *same binary* with dif
| --- | --- |
| RPC secret | Internode RPC fails closed: remote endpoints with default credentials and no `RUSTFS_RPC_SECRET` abort startup with `store init aborted: endpoints include remote nodes but ...` (`crates/ecstore/src/store/init.rs`). Set a non-default `RUSTFS_RPC_SECRET`, identical on every node. |
| systemd start timeout | `deploy/build/rustfs.service` is `Type=notify` and ships `TimeoutStartSec=120s`; READY fires only after quorum. If freshly purged disks need longer, raise it in a drop-in rather than lowering it. |
| Metrics export | RustFS has no Prometheus pull endpoint (`/admin/v3/metrics` is NDJSON, see `rustfs/src/admin/handlers/metrics.rs`); it pushes OTLP. Run an otel-collector (OTLP receiver → Prometheus exporter) and set `RUSTFS_OBS_ENDPOINT`, `RUSTFS_OBS_METRICS_EXPORT_ENABLED=true`, `RUSTFS_OBS_METER_INTERVAL=5`. For lock p99 also set `RUSTFS_OBJECT_LOCK_DIAG_ENABLE=true` (default off). |
| Metrics export | RustFS has no Prometheus pull endpoint (`/admin/v3/realtime` is NDJSON, see `rustfs/src/admin/handlers/realtime.rs`); it pushes OTLP. Run an otel-collector (OTLP receiver → Prometheus exporter) and set `RUSTFS_OBS_ENDPOINT`, `RUSTFS_OBS_METRICS_EXPORT_ENABLED=true`, `RUSTFS_OBS_METER_INTERVAL=5`. For lock p99 also set `RUSTFS_OBJECT_LOCK_DIAG_ENABLE=true` (default off). |
| Server env | `RUSTFS_INTERNODE_*` are **server** env. For p0/p1/p2, source the emitted `server-env.sh` on every node and restart before the run; the driver cannot mutate a running server. |
## Driver
@@ -20,7 +20,7 @@ sum by (instance) (rate(rustfs_s3_http_requests_total{method="PUT",outcome=~"[1-
Inspect `service_error` and `cancelled` separately; neither implies a received HTTP status. A zero denominator or absent series means no observed traffic, not proof of health. Use `rate` or reset-aware deltas because counters restart with the process. The older `rustfs_s3_operations_total` counter measures handler entries and excludes pre-dispatch rejections; it is not this HTTP denominator.
The authenticated admin metrics endpoint exposes the same counters through the optional `http` field in `aggregated` and `by_host`. Request `/rustfs/admin/v3/metrics?types=512&by-host=true&n=1` on each node for HTTP-only data. The default type selection also includes HTTP outcomes. This endpoint remains an NDJSON stream and does not become a cluster-wide peer fanout. `http.requests` contains `method`, `operation`, `outcome`, and `total`; `http.collected` timestamps collection. Compare consecutive samples from the same host. Concurrent snapshots are not atomic across series.
The authenticated admin metrics endpoint exposes the same counters through the optional `http` field in `aggregated` and `by_host`. Request `/rustfs/admin/v3/realtime?types=512&by-host=true&n=1` on each node for HTTP-only data. The default type selection also includes HTTP outcomes. This endpoint remains an NDJSON stream and does not become a cluster-wide peer fanout. `http.requests` contains `method`, `operation`, `outcome`, and `total`; `http.collected` timestamps collection. Compare consecutive samples from the same host. Concurrent snapshots are not atomic across series.
A missing `http` field means an older or non-reporting node, not zero failures. Old map-encoded RPC readers ignore the additive field; new readers accept older snapshots. In mixed-version deployments, check reporting coverage before aggregating a fleet-wide ratio.
+1 -1
View File
@@ -317,7 +317,7 @@ Use the harness around a post-start bucket creation workload to cover the timing
1. Start RustFS from an empty data path.
2. Start the harness before creating buckets.
3. Create a bucket, upload objects, and keep the harness running until at least one usage save is observed.
4. Compare `scanner-summary.csv` with `/rustfs/admin/v3/metrics?types=1&n=1` bucket metrics.
4. Compare `scanner-summary.csv` with `/rustfs/admin/v3/realtime?types=1&n=1` bucket metrics.
Expected evidence: dirty usage is marked, `life_time_scan_cycle` or `life_time_scan_bucket_drive` advances, `life_time_scan_object` advances for object workloads, and `life_time_save_usage` plus `usage_last_save_result=success` appear before non-zero bucket usage metrics are accepted as fresh.
+2 -2
View File
@@ -481,7 +481,7 @@ Read this snapshot before changing scanner controls: `blocked_source` with `life
## Reading Distributed Metrics
`/rustfs/admin/v3/scanner/status` and `/rustfs/admin/v3/metrics` report the node that handles the HTTP request; the metrics endpoint does not fan out to peers. In distributed deployments, query every node explicitly and keep `by-host=true` so each response includes that node's host view:
`/rustfs/admin/v3/scanner/status` and `/rustfs/admin/v3/realtime` report the node that handles the HTTP request; the metrics endpoint does not fan out to peers. In distributed deployments, query every node explicitly and keep `by-host=true` so each response includes that node's host view:
```bash
for endpoint in http://node-a:9000 http://node-b:9000 http://node-c:9000; do
@@ -493,7 +493,7 @@ for endpoint in http://node-a:9000 http://node-b:9000 http://node-c:9000; do
--access_key "$RUSTFS_ACCESS_KEY" \
--secret_key "$RUSTFS_SECRET_KEY" \
--request GET \
"${endpoint}/rustfs/admin/v3/metrics?types=1&by-host=true&n=1" \
"${endpoint}/rustfs/admin/v3/realtime?types=1&by-host=true&n=1" \
> "artifacts/scanner-metrics.${node}.$(date -u +%Y%m%dT%H%M%SZ).ndjson"
done
```
@@ -189,6 +189,7 @@ mod tests {
use crate::admin::handlers::target_descriptor::admin_target_spec_from_builtin;
use crate::admin::runtime_sources::{IamInterface, KmsInterface};
use crate::admin::storage_api::config::save_admin_server_config;
use crate::admin::storage_api::error::StorageError;
use rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS;
use rustfs_config::server_config::KVS;
use rustfs_config::{ENABLE_KEY, EnableState, SCANNER_CYCLE, SCANNER_SUB_SYS, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR};
@@ -231,11 +232,17 @@ mod tests {
let mut poll = tokio::time::interval(Duration::from_millis(10));
loop {
poll.tick().await;
let config = read_admin_config_without_migrate(store.clone())
.await
.expect("read persisted server config");
if config.0.get(subsystem).is_some_and(|targets| targets.contains_key(target)) {
return;
match read_admin_config_without_migrate(store.clone()).await {
Ok(config) => {
if config.0.get(subsystem).is_some_and(|targets| targets.contains_key(target)) {
return;
}
}
Err(StorageError::Lock(rustfs_lock::LockError::Timeout { .. })) => {
// The writer may still be committing the config snapshot. Retry after
// the object lock is released instead of failing the polling helper.
}
Err(error) => panic!("read persisted server config: {error}"),
}
}
})
+2 -2
View File
@@ -43,7 +43,6 @@ pub mod kms_key_metadata;
pub mod kms_keys;
pub mod kms_management;
pub mod kms_rekey;
pub mod metrics;
pub mod mfa;
pub mod module_switch;
mod notify_runtime_access;
@@ -58,6 +57,7 @@ pub mod pools;
pub mod profile;
pub mod profile_admin;
pub mod quota;
pub mod realtime;
pub mod rebalance;
pub mod replication;
pub mod scanner;
@@ -145,7 +145,7 @@ mod tests {
let _inspect_data_handler = system::InspectDataHandler {};
let _storage_info_handler = system::StorageInfoHandler {};
let _data_usage_handler = system::DataUsageInfoHandler {};
let _metrics_handler = metrics::MetricsHandler {};
let _metrics_handler = realtime::MetricsHandler {};
let _profile_handler = profile_admin::ProfileHandler {};
let _profile_status_handler = profile_admin::ProfileStatusHandler {};
let _tls_status_handler = tls_debug::TlsStatusHandler {};
@@ -14,7 +14,7 @@
//! Console realtime metrics API.
//!
//! This preserves the console's fixed `/admin/v3/metrics` contract while
//! This preserves the console's fixed `/admin/v3/realtime` contract while
//! keeping the response format explicitly NDJSON. It is not a Prometheus text
//! exposition endpoint.
@@ -302,7 +302,7 @@ mod tests {
#[test]
fn metrics_params_default_to_single_sample() {
let uri: Uri = "/rustfs/admin/v3/metrics".parse().unwrap();
let uri: Uri = "/rustfs/admin/v3/realtime".parse().unwrap();
let mp = extract_metrics_init_params(&uri);
assert_eq!(resolve_sample_count(&mp), DEFAULT_METRICS_SAMPLES);
@@ -310,7 +310,7 @@ mod tests {
#[test]
fn metrics_params_treat_zero_as_single_sample() {
let uri: Uri = "/rustfs/admin/v3/metrics?n=0".parse().unwrap();
let uri: Uri = "/rustfs/admin/v3/realtime?n=0".parse().unwrap();
let mp = extract_metrics_init_params(&uri);
assert_eq!(resolve_sample_count(&mp), DEFAULT_METRICS_SAMPLES);
@@ -318,7 +318,7 @@ mod tests {
#[test]
fn metrics_params_cap_samples_to_safety_limit() {
let uri: Uri = "/rustfs/admin/v3/metrics?n=9999".parse().unwrap();
let uri: Uri = "/rustfs/admin/v3/realtime?n=9999".parse().unwrap();
let mp = extract_metrics_init_params(&uri);
assert_eq!(resolve_sample_count(&mp), MAX_METRICS_SAMPLES);
@@ -332,7 +332,7 @@ mod tests {
#[tokio::test]
async fn metrics_handler_rejects_missing_credentials() {
let result = MetricsHandler {}
.call(build_metrics_request("/rustfs/admin/v3/metrics"), Params::new())
.call(build_metrics_request("/rustfs/admin/v3/realtime"), Params::new())
.await;
let err = match result {
Ok(_) => panic!("metrics handler must reject unauthenticated requests"),
+222 -16
View File
@@ -43,13 +43,21 @@ use s3s::{
s3_error,
};
use serde::{Deserialize, Serialize};
use std::{sync::Arc, time::Duration};
use std::{
future::Future,
sync::Arc,
time::{Duration, Instant},
};
use time::OffsetDateTime;
use tracing::{error, info, warn};
const LOG_COMPONENT_ADMIN: &str = "admin";
const LOG_SUBSYSTEM_REBALANCE: &str = "rebalance";
const EVENT_ADMIN_REBALANCE_STATE: &str = "admin_rebalance_state";
const REBALANCE_START_FLEET_PROOF_MARKER: &str = "pool activation requires a live fleet capability proof";
const REBALANCE_START_FLEET_PROOF_RETRY_BUDGET: Duration = Duration::from_secs(30);
const REBALANCE_START_FLEET_PROOF_RETRY_DELAY: Duration = Duration::from_secs(5);
const REBALANCE_START_RETRY_AFTER_SECS: &str = "5";
fn admin_request_id(headers: &HeaderMap) -> Option<&str> {
headers
@@ -105,6 +113,80 @@ fn rebalance_internal_error(message: impl Into<String>) -> S3Error {
S3Error::with_message(S3ErrorCode::InternalError, message.into())
}
fn rebalance_fleet_proof_unavailable_error() -> S3Error {
let mut err = S3Error::with_message(
S3ErrorCode::ServiceUnavailable,
"rebalance start is waiting for cluster capability readiness; retry later",
);
let mut headers = HeaderMap::new();
headers.insert(http::header::RETRY_AFTER, HeaderValue::from_static(REBALANCE_START_RETRY_AFTER_SECS));
err.set_headers(headers);
err
}
fn is_rebalance_start_fleet_proof_retryable(err: &StorageError) -> bool {
crate::storage_api::capacity::is_pool_activation_fleet_proof_error(err)
&& err.to_string().contains(REBALANCE_START_FLEET_PROOF_MARKER)
}
fn rebalance_start_failure_error(start_err: &StorageError, rollback_result: &Result<(), String>) -> S3Error {
if crate::storage_api::capacity::is_pool_activation_fleet_proof_error(start_err) {
if rollback_result.is_ok() {
return rebalance_fleet_proof_unavailable_error();
}
return rebalance_internal_error("failed to roll back rebalance start after fleet capability readiness failure");
}
rebalance_internal_error(rebalance_start_rollback_error(&start_err.to_string(), rollback_result))
}
async fn retry_rebalance_start_operation<T, F, Fut>(
operation: &'static str,
rebalance_id: &str,
request_id: &str,
actor: &str,
remote_addr: &str,
deadline: Instant,
mut operation_fn: F,
) -> Result<T, StorageError>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, StorageError>>,
{
let mut attempt = 1_u32;
loop {
match operation_fn().await {
Ok(value) => return Ok(value),
Err(err) if is_rebalance_start_fleet_proof_retryable(&err) => {
let now = Instant::now();
if now >= deadline {
return Err(err);
}
let delay = REBALANCE_START_FLEET_PROOF_RETRY_DELAY.min(deadline.saturating_duration_since(now));
warn!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REBALANCE,
action = "start",
state = "fleet_proof_retry_scheduled",
result = "retrying",
operation,
attempt,
retry_delay_ms = delay.as_millis(),
request_id = %request_id,
actor = %actor,
remote_addr = %remote_addr,
rebalance_id = %rebalance_id,
"admin rebalance state"
);
tokio::time::sleep(delay).await;
attempt = attempt.saturating_add(1);
}
Err(err) => return Err(err),
}
}
}
fn rebalance_rollback_stop_failure_message(rebalance_id: &str, failures: &[String]) -> String {
format!("cluster stop_rebalance rollback for {rebalance_id} partial: {}", failures.join("; "))
}
@@ -209,7 +291,7 @@ async fn rollback_rebalance_start_for_admin(
store: &Arc<ECStore>,
notification_sys: Option<&NotificationSys>,
rebalance_id: &str,
start_err: &str,
start_err: &StorageError,
request_id: &str,
actor: &str,
remote_addr: &str,
@@ -246,7 +328,7 @@ async fn rollback_rebalance_start_for_admin(
),
}
Err(rebalance_internal_error(rebalance_start_rollback_error(start_err, &rollback_result)))
Err(rebalance_start_failure_error(start_err, &rollback_result))
}
pub fn register_rebalance_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
@@ -562,6 +644,7 @@ impl Operation for RebalanceStart {
"admin rebalance state"
);
let notification_sys = current_notification_system();
let fleet_proof_deadline = Instant::now() + REBALANCE_START_FLEET_PROOF_RETRY_BUDGET;
for step in rebalance_start_steps(notification_sys.is_some()) {
match step {
RebalanceStartStep::PropagateFence => {
@@ -592,13 +675,11 @@ impl Operation for RebalanceStart {
error = %err,
"admin rebalance state"
);
let start_err = err.to_string();
rollback_rebalance_start_for_admin(
&store,
Some(notification_sys),
&id,
&start_err,
&err,
&request_id,
&actor,
&remote_addr,
@@ -608,7 +689,17 @@ impl Operation for RebalanceStart {
}
}
RebalanceStartStep::StartLocal => {
if let Err(err) = store.start_rebalance_for_id(&id).await {
let start_result = retry_rebalance_start_operation(
"start_rebalance_for_id",
&id,
&request_id,
&actor,
&remote_addr,
fleet_proof_deadline,
|| store.start_rebalance_for_id(&id),
)
.await;
if let Err(err) = start_result {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
@@ -682,6 +773,9 @@ impl Operation for RebalanceStart {
)));
}
}
if crate::storage_api::capacity::is_pool_activation_fleet_proof_error(&err) {
return Err(rebalance_fleet_proof_unavailable_error());
}
return Err(rebalance_internal_error(format!(
"failed to start rebalance after metadata initialized for {id}; local metadata was finalized as failed: {start_err}"
)));
@@ -701,7 +795,17 @@ impl Operation for RebalanceStart {
rebalance_id = %id,
"admin rebalance state"
);
if let Err(err) = notification_sys.load_rebalance_meta(true).await {
let worker_result = retry_rebalance_start_operation(
"load_rebalance_meta(start=true)",
&id,
&request_id,
&actor,
&remote_addr,
fleet_proof_deadline,
|| notification_sys.load_rebalance_meta(true),
)
.await;
if let Err(err) = worker_result {
error!(
event = EVENT_ADMIN_REBALANCE_STATE,
component = LOG_COMPONENT_ADMIN,
@@ -715,13 +819,11 @@ impl Operation for RebalanceStart {
error = %err,
"admin rebalance state"
);
let start_err = err.to_string();
rollback_rebalance_start_for_admin(
&store,
Some(notification_sys),
&id,
&start_err,
&err,
&request_id,
&actor,
&remote_addr,
@@ -1059,12 +1161,14 @@ mod rebalance_handler_tests {
use super::calculate_rebalance_progress;
use super::{
Body, HeaderMap, Method, Operation, Params, RebalPoolProgress, RebalanceAdminStatus, RebalancePoolStatus, RebalanceStart,
RebalanceStartStep, RebalanceStatus, RebalanceStop, RebalanceStopPropagationStatus, S3ErrorCode, S3Request, Uri,
build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status,
rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets, rebalance_rollback_failure_message,
rebalance_rollback_stop_failure_message, rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id,
rebalance_used_pct, rollback_result_label, stop_rebalance_admission_first,
RebalanceStartStep, RebalanceStatus, RebalanceStop, RebalanceStopPropagationStatus, S3ErrorCode, S3Request, StatusCode,
Uri, build_rebalance_admin_status, build_rebalance_pool_statuses, build_rebalance_stop_propagation_status,
is_rebalance_start_fleet_proof_retryable, rebalance_pool_used, rebalance_query_present, rebalance_remaining_buckets,
rebalance_rollback_failure_message, rebalance_rollback_stop_failure_message, rebalance_start_failure_error,
rebalance_start_rollback_error, rebalance_start_steps, rebalance_stop_target_id, rebalance_used_pct,
retry_rebalance_start_operation, rollback_result_label, stop_rebalance_admission_first,
};
use crate::admin::storage_api::error::StorageError;
use crate::admin::storage_api::rebalance::{
DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo,
RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord, encode_rebalance_stop_propagation_record,
@@ -1328,6 +1432,108 @@ mod rebalance_handler_tests {
assert!(message.contains("rollback error: peer b stop_rebalance failed: timeout"));
}
#[test]
fn test_rebalance_start_fleet_proof_maps_to_retryable_503() {
let start_err = StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER);
let err = rebalance_start_failure_error(&start_err, &Ok(()));
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
assert_eq!(err.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
assert_eq!(
err.headers()
.and_then(|headers| headers.get(http::header::RETRY_AFTER))
.and_then(|value| value.to_str().ok()),
Some(super::REBALANCE_START_RETRY_AFTER_SECS)
);
assert!(
!err.message()
.expect("retryable rebalance start error should have a message")
.contains(super::REBALANCE_START_FLEET_PROOF_MARKER)
);
}
#[test]
fn test_rebalance_start_fleet_proof_rollback_failure_is_generic_internal_error() {
let start_err = StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER);
let rollback_result = Err("peer rollback failed".to_string());
let err = rebalance_start_failure_error(&start_err, &rollback_result);
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert!(
!err.message()
.expect("internal rebalance start error should have a message")
.contains(super::REBALANCE_START_FLEET_PROOF_MARKER)
);
}
#[test]
fn test_rebalance_start_unrelated_failure_remains_internal_error() {
let start_err = StorageError::other("disk read failed");
let err = rebalance_start_failure_error(&start_err, &Ok(()));
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert!(err.message().is_some_and(|message| message.contains("disk read failed")));
}
#[test]
fn test_rebalance_start_fleet_proof_retry_only_matches_missing_live_proof() {
let missing = StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER);
let expired = StorageError::other(format!(
"rebalance meta save failed during start_rebalance: {}",
"pool activation fleet capability proof expired before commit"
));
assert!(is_rebalance_start_fleet_proof_retryable(&missing));
assert!(!is_rebalance_start_fleet_proof_retryable(&expired));
}
#[tokio::test(start_paused = true)]
async fn test_rebalance_start_retry_waits_for_fleet_proof() {
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
let attempts = Arc::new(AtomicUsize::new(0));
let task = tokio::spawn({
let attempts = Arc::clone(&attempts);
async move {
retry_rebalance_start_operation(
"load_rebalance_meta(start=true)",
"rebalance-id",
"request-id",
"actor",
"remote",
std::time::Instant::now() + super::REBALANCE_START_FLEET_PROOF_RETRY_BUDGET,
move || {
let attempts = Arc::clone(&attempts);
async move {
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
if attempt == 0 {
Err(StorageError::other(super::REBALANCE_START_FLEET_PROOF_MARKER))
} else {
Ok(())
}
}
},
)
.await
}
});
tokio::task::yield_now().await;
assert_eq!(attempts.load(Ordering::SeqCst), 1);
tokio::time::advance(super::REBALANCE_START_FLEET_PROOF_RETRY_DELAY).await;
task.await
.expect("retry task should not panic")
.expect("fleet proof retry should eventually succeed");
assert_eq!(attempts.load(Ordering::SeqCst), 2);
}
#[test]
fn test_rebalance_rollback_stop_failure_message_lists_failures() {
let failures = vec![
+3 -3
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::{cluster_snapshot, metrics};
use super::{cluster_snapshot, realtime};
use crate::admin::auth::authorize_admin_request;
use crate::admin::handlers::account::{ACCOUNT_INFO_ROUTE, ACCOUNT_PASSWORD_ROUTE};
use crate::admin::handlers::mfa::{ACCOUNT_MFA_ROUTE, MFA_CHALLENGE_ROUTE, USER_MFA_ROUTE};
@@ -167,8 +167,8 @@ pub fn register_system_route(r: &mut S3Router<AdminOperation>) -> std::io::Resul
r.insert(
Method::GET,
format!("{}{}", ADMIN_PREFIX, "/v3/metrics").as_str(),
AdminOperation(&metrics::MetricsHandler {}),
format!("{}{}", ADMIN_PREFIX, "/v3/realtime").as_str(),
AdminOperation(&realtime::MetricsHandler {}),
)?;
r.insert(
+2 -2
View File
@@ -320,7 +320,7 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
),
admin(HttpMethod::Get, "/rustfs/admin/v3/info", SERVER_INFO, RouteRiskLevel::Sensitive),
admin(HttpMethod::Get, "/rustfs/admin/v3/storageinfo", STORAGE_INFO, RouteRiskLevel::Sensitive),
admin(HttpMethod::Get, "/rustfs/admin/v3/metrics", GET_METRICS, RouteRiskLevel::Sensitive),
admin(HttpMethod::Get, "/rustfs/admin/v3/realtime", GET_METRICS, RouteRiskLevel::Sensitive),
admin(
HttpMethod::Get,
"/rustfs/admin/v3/object-data-cache/stats",
@@ -2303,7 +2303,7 @@ mod tests {
#[test]
fn route_policy_maps_metrics_to_explicit_admin_action() {
assert_action(HttpMethod::Get, "/rustfs/admin/v3/metrics", GET_METRICS);
assert_action(HttpMethod::Get, "/rustfs/admin/v3/realtime", GET_METRICS);
}
#[test]
+2 -2
View File
@@ -190,7 +190,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
admin_route(Method::GET, "/v3/storageinfo"),
admin_route(Method::GET, "/v3/datausageinfo"),
admin_route_sample(Method::GET, "/v3/usage/{bucket}", "/v3/usage/test-bucket"),
admin_route(Method::GET, "/v3/metrics"),
admin_route(Method::GET, "/v3/realtime"),
admin_route(Method::GET, "/v3/object-data-cache/stats"),
admin_route(Method::POST, "/v3/object-data-cache/flush"),
admin_route(Method::GET, "/v3/pools/list"),
@@ -1333,7 +1333,7 @@ fn test_register_routes_cover_representative_admin_paths() {
assert_route(&router, Method::POST, &admin_path("/v3/service"));
assert_route(&router, Method::GET, &admin_path("/v3/info"));
assert_route(&router, Method::GET, &admin_path("/v3/storageinfo"));
assert_route(&router, Method::GET, &admin_path("/v3/metrics"));
assert_route(&router, Method::GET, &admin_path("/v3/realtime"));
assert_route(&router, Method::GET, &admin_path("/v3/pools/list"));
assert_route(&router, Method::GET, &admin_path("/v3/decommission/status"));
+1 -1
View File
@@ -2538,7 +2538,7 @@ mod tests {
assert!(futures::poll!(cancelled.as_mut()).is_pending());
drop(cancelled);
for path in [
"/rustfs/admin/v3/metrics",
"/rustfs/admin/v3/realtime",
"/minio/admin/v3/storageinfo",
CONSOLE_PREFIX,
"/rustfs/rpc/test",
+7 -4
View File
@@ -1712,16 +1712,17 @@ fn response_with_status(status: StatusCode, message: impl Into<String>) -> Respo
}
fn response_with_disk_error(error: &DiskError, message: impl Into<String>) -> Response<Body> {
let missing = match error {
let disk_error = match error {
DiskError::FileNotFound => Some(rustfs_rio::INTERNODE_FILE_NOT_FOUND),
DiskError::VolumeNotFound => Some(rustfs_rio::INTERNODE_VOLUME_NOT_FOUND),
DiskError::FileCorrupt => Some(rustfs_rio::INTERNODE_FILE_CORRUPT),
_ => None,
};
let mut response = response_with_status(StatusCode::INTERNAL_SERVER_ERROR, message);
if let Some(missing) = missing {
if let Some(disk_error) = disk_error {
response
.headers_mut()
.insert(rustfs_rio::INTERNODE_DISK_ERROR_HEADER, HeaderValue::from_static(missing));
.insert(rustfs_rio::INTERNODE_DISK_ERROR_HEADER, HeaderValue::from_static(disk_error));
}
response
}
@@ -3087,12 +3088,14 @@ mod tests {
}
#[test]
fn read_file_error_response_marks_only_missing_disk_errors() {
fn read_file_error_response_preserves_typed_disk_errors() {
for (error, expected) in [
(DiskError::FileNotFound, rustfs_rio::INTERNODE_FILE_NOT_FOUND),
(DiskError::VolumeNotFound, rustfs_rio::INTERNODE_VOLUME_NOT_FOUND),
(DiskError::FileCorrupt, rustfs_rio::INTERNODE_FILE_CORRUPT),
] {
let response = response_with_disk_error(&error, error.to_string());
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
assert_eq!(
response.headers().get(rustfs_rio::INTERNODE_DISK_ERROR_HEADER),
Some(&HeaderValue::from_static(expected))
+97 -4
View File
@@ -52,9 +52,11 @@ use serde::Deserialize;
use sha2::{Digest, Sha256};
use std::{
collections::HashMap,
future::Future,
io::Cursor,
pin::Pin,
sync::{Arc, LazyLock, OnceLock},
time::Instant,
};
use time::OffsetDateTime;
use tokio::spawn;
@@ -78,6 +80,9 @@ const EVENT_RPC_REQUEST_FAILED: &str = "rpc_request_failed";
const EVENT_RPC_RESPONSE_EMITTED: &str = "rpc_response_emitted";
const EVENT_RPC_BACKGROUND_TASK_SPAWNED: &str = "rpc_background_task_spawned";
const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed";
const REBALANCE_START_FLEET_PROOF_MARKER: &str = "pool activation requires a live fleet capability proof";
const REBALANCE_START_FLEET_PROOF_RETRY_BUDGET: Duration = Duration::from_secs(30);
const REBALANCE_START_FLEET_PROOF_RETRY_DELAY: Duration = Duration::from_secs(5);
const HEAL_CONTROL_REPLAY_CACHE_MAX_ENTRIES: usize = 4096;
const TIER_MUTATION_PEER_STATE_UNSPECIFIED_WIRE: i32 = 0;
const TIER_MUTATION_PEER_STATE_PREPARED_WIRE: i32 = 1;
@@ -501,6 +506,48 @@ fn background_rebalance_start_error_message(result: StorageResult<()>) -> Option
result.err().map(|err| format!("start_rebalance failed: {err}"))
}
fn is_rebalance_start_fleet_proof_retryable(err: &Error) -> bool {
crate::storage::storage_api::ecstore_capacity::is_pool_activation_fleet_proof_error(err)
&& err.to_string().contains(REBALANCE_START_FLEET_PROOF_MARKER)
}
async fn retry_rebalance_start_fleet_proof<F, Fut>(mut operation: F) -> StorageResult<()>
where
F: FnMut() -> Fut,
Fut: Future<Output = StorageResult<()>>,
{
let deadline = Instant::now() + REBALANCE_START_FLEET_PROOF_RETRY_BUDGET;
let mut attempt = 1_u32;
loop {
match operation().await {
Ok(()) => return Ok(()),
Err(err) if is_rebalance_start_fleet_proof_retryable(&err) => {
let now = Instant::now();
if now >= deadline {
return Err(err);
}
let delay = REBALANCE_START_FLEET_PROOF_RETRY_DELAY.min(deadline.saturating_duration_since(now));
warn!(
event = EVENT_RPC_BACKGROUND_TASK_FAILED,
component = LOG_COMPONENT_STORAGE,
subsystem = LOG_SUBSYSTEM_REBALANCE,
operation = "start_rebalance",
state = "fleet_proof_retry_scheduled",
result = "retrying",
attempt,
retry_delay_ms = delay.as_millis(),
error = %err,
"node rpc background task retry"
);
tokio::time::sleep(delay).await;
attempt = attempt.saturating_add(1);
}
Err(err) => return Err(err),
}
}
}
fn stop_rebalance_response(result: StorageResult<()>) -> StopRebalanceResponse {
match result {
Ok(_) => StopRebalanceResponse {
@@ -2800,7 +2847,9 @@ impl Node for NodeService {
if start_rebalance {
log_background_rebalance_task_spawned!(start_rebalance);
if let Some(message) = background_rebalance_start_error_message(store.start_rebalance().await) {
if let Some(message) =
background_rebalance_start_error_message(retry_rebalance_start_fleet_proof(|| store.start_rebalance()).await)
{
error!(
event = EVENT_RPC_BACKGROUND_TASK_FAILED,
component = LOG_COMPONENT_STORAGE,
@@ -2975,9 +3024,10 @@ mod tests {
SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_PUBLICATION_LEASE_TTL_MS,
SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, STORAGE_CLASS_SUB_SYS, admit_heal_control_replay,
background_rebalance_start_error_message, execute_heal_control_envelope_with_manager,
initialize_heal_topology_fingerprint, initialize_heal_topology_fingerprint_with_probe, legacy_scanner_activity_response,
make_heal_control_server, make_heal_control_server_with_cache, make_server, make_server_for_context,
make_tier_mutation_control_server_for_context, previous_scanner_activity_response, remove_heal_control_replay,
initialize_heal_topology_fingerprint, initialize_heal_topology_fingerprint_with_probe,
is_rebalance_start_fleet_proof_retryable, legacy_scanner_activity_response, make_heal_control_server,
make_heal_control_server_with_cache, make_server, make_server_for_context, make_tier_mutation_control_server_for_context,
previous_scanner_activity_response, remove_heal_control_replay, retry_rebalance_start_fleet_proof,
scanner_activity_response_v7, start_decommission_failure_response, stop_rebalance_response,
validate_admin_heal_control_start,
};
@@ -8121,6 +8171,49 @@ mod tests {
assert!(message.contains("boom"));
}
#[test]
fn test_rebalance_start_retry_ignores_expired_fleet_proof() {
let expired = Error::other("pool activation fleet capability proof expired before commit");
assert!(!is_rebalance_start_fleet_proof_retryable(&expired));
}
#[tokio::test(start_paused = true)]
async fn test_retry_rebalance_start_waits_for_fleet_proof() {
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
let attempts = Arc::new(AtomicUsize::new(0));
let task = tokio::spawn({
let attempts = Arc::clone(&attempts);
async move {
retry_rebalance_start_fleet_proof(move || {
let attempts = Arc::clone(&attempts);
async move {
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
if attempt == 0 {
Err(Error::other(super::REBALANCE_START_FLEET_PROOF_MARKER))
} else {
Ok(())
}
}
})
.await
}
});
tokio::task::yield_now().await;
assert_eq!(attempts.load(Ordering::SeqCst), 1);
tokio::time::advance(super::REBALANCE_START_FLEET_PROOF_RETRY_DELAY).await;
task.await
.expect("retry task should not panic")
.expect("fleet proof retry should eventually succeed");
assert_eq!(attempts.load(Ordering::SeqCst), 2);
}
#[test]
fn test_stop_rebalance_response_reports_local_stop_error() {
let response = stop_rebalance_response(Err(Error::other("boom")));
@@ -2277,7 +2277,7 @@ fi
rustfs/src/admin/router.rs \
rustfs/src/admin/console.rs \
rustfs/src/admin/handlers/heal.rs \
rustfs/src/admin/handlers/metrics.rs \
rustfs/src/admin/handlers/realtime.rs \
rustfs/src/admin/handlers/object_zip_download.rs || true
) >"$RUSTFS_APP_ADMIN_STORAGE_API_BYPASS_HITS_FILE"
+1 -1
View File
@@ -50,7 +50,7 @@
1|crates/ecstore/src/services/rebalance/entry.rs
8|crates/ecstore/src/services/rebalance/meta.rs
8|crates/ecstore/src/services/rebalance/runtime.rs
18|crates/ecstore/src/services/rebalance/worker.rs
17|crates/ecstore/src/services/rebalance/worker.rs
33|crates/ecstore/src/services/tier/tier.rs
1|crates/ecstore/src/services/tier/tier_config.rs
1|crates/ecstore/src/services/tier/warm_backend.rs