From 0c3617f3f82a2e6eee299d6a5f0b4bb4531dd252 Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 30 Sep 2026 08:04:13 +0800 Subject: [PATCH] fix(admin): route MRF errors through gateway facade Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/mrf_queue.rs | 4 +- .../heal/src/heal/mrf_queue/partial_write.rs | 105 ++++++++++-------- crates/heal/tests/mrf_partial_write_test.rs | 22 +++- rustfs/src/admin/handlers/heal.rs | 64 ++++++----- 4 files changed, 120 insertions(+), 75 deletions(-) diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 2769bf7fb..77bdba8db 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -948,7 +948,9 @@ impl MrfRuntime { intents .iter() .map(|(intent, record)| match record { - Some(record) => PartialWrites::cost_with_state(intent, &record.state), + Some(record) => { + PartialWrites::cost_with_state_and_audit(intent, &record.state, record.last_operator_acceptance.as_ref()) + } None => PartialWrites::cost(intent), }) .sum::(), diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index 781b5973f..11fa699d8 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -190,6 +190,19 @@ impl PartialWrites { + state.estimated_bytes() } + pub(super) fn cost_with_state_and_audit( + intent: &MrfIntent, + state: &ResponsibilityState, + audit: Option<&MrfOperatorAcceptance>, + ) -> usize { + Self::cost_with_state(intent, state) + + audit.map_or(0, |audit| audit.actor.len() + audit.reason.len() + audit.reference.len()) + } + + fn entry_cost(entry: &Responsibility) -> usize { + Self::cost_with_state_and_audit(&entry.intent, &entry.state, entry.last_operator_acceptance.as_ref()) + } + #[cfg(test)] pub(super) fn admit( &mut self, @@ -217,7 +230,7 @@ impl PartialWrites { let previous = self.entries.get(&key); let previous_was_held = previous.is_some_and(|entry| entry.state.is_parked()); let mut retry_queued = previous.is_some_and(|entry| entry.retry_queued); - let old_cost = previous.map_or(0, |entry| Self::cost_with_state(&entry.intent, &entry.state)); + let old_cost = previous.map_or(0, Self::entry_cost); let next_bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::cost(&intent)); if (previous.is_none() && self.entries.len() >= capacity) || next_bytes > byte_budget { return Err(MrfDurableAdmissionError::Full); @@ -286,15 +299,12 @@ impl PartialWrites { { return false; } - let old_cost = Self::cost_with_state(&entry.intent, &entry.state); + let old_cost = Self::entry_cost(entry); entry.state = ResponsibilityState::HeldUnverifiedLegacy { bucket_incarnation_id: anchor.bucket_incarnation_id, since_ms: unix_now_ms(), }; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); true } @@ -372,15 +382,12 @@ impl PartialWrites { if !matches!(&entry.state, ResponsibilityState::Active) { return false; } - let old_cost = Self::cost_with_state(&entry.intent, &entry.state); + let old_cost = Self::entry_cost(entry); entry.responsibility_id = record.responsibility_id; entry.source_bucket_incarnation_id = record.source_bucket_incarnation_id; entry.last_operator_acceptance = record.last_operator_acceptance.clone(); entry.state = record.state.clone(); - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); if entry.state.is_parked() { entry.retry_queued = false; } @@ -480,7 +487,7 @@ impl PartialWrites { } let previous = entry.state.clone(); let previous_acceptance = entry.last_operator_acceptance.clone(); - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); let accepted_at_ms = unix_now_ms(); let acceptance = MrfOperatorAcceptance { accepted_at_ms, @@ -501,7 +508,7 @@ impl PartialWrites { reference, request_id, }; - let next_cost = Self::cost_with_state(&entry.intent, &next_state); + let next_cost = Self::cost_with_state_and_audit(&entry.intent, &next_state, Some(&acceptance)); let next_bytes = self.bytes.saturating_sub(old_cost).saturating_add(next_cost); if next_bytes > byte_budget { return Err("durable responsibility byte budget is exhausted"); @@ -544,9 +551,9 @@ impl PartialWrites { return Err("bucket incarnation changed; refresh the responsibility listing"); } let previous = entry.state.clone(); - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); entry.state = ResponsibilityState::Active; - self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::cost(&entry.intent)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.persisted = false; entry.next_attempt = Instant::now(); if !entry.retry_queued { @@ -572,13 +579,10 @@ impl PartialWrites { else { return false; }; - let old_cost = Self::cost_with_state(&entry.intent, &entry.state); + let old_cost = Self::entry_cost(entry); entry.state = state; entry.last_operator_acceptance = last_operator_acceptance; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.persisted = true; if entry.state.is_parked() { entry.retry_queued = false; @@ -675,7 +679,7 @@ impl PartialWrites { let previous = entry.state.clone(); let previous_acceptance = entry.last_operator_acceptance.clone(); if source_bucket_incarnation_id.is_some_and(|source| source != observed_bucket_incarnation_id) { - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); entry.state = ResponsibilityState::BucketIncarnationChanged { source_bucket_incarnation_id: match source_bucket_incarnation_id { Some(source) => source, @@ -684,21 +688,15 @@ impl PartialWrites { observed_bucket_incarnation_id, detected_at_ms: unix_now_ms(), }; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.retry_queued = false; } else if source_bucket_incarnation_id.is_none() { - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); entry.state = ResponsibilityState::LegacyGenerationUnknown { observed_bucket_incarnation_id, detected_at_ms: unix_now_ms(), }; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.retry_queued = false; } Ok((previous, previous_acceptance)) @@ -727,8 +725,7 @@ impl PartialWrites { entry.anchor = manager.durable_mrf_repair_anchor(&entry.intent).await; if let Some(anchor) = entry.anchor.clone() { if entry.source_bucket_incarnation_id.is_none() { - let previous = entry.state.clone(); - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); let detected_at_ms = SystemTime::now() .duration_since(UNIX_EPOCH) .map_or(0, |duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)); @@ -736,10 +733,7 @@ impl PartialWrites { observed_bucket_incarnation_id: anchor.bucket_incarnation_id, detected_at_ms, }; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.retry_queued = false; lifecycle_changed = true; continue; @@ -747,8 +741,7 @@ impl PartialWrites { if let Some(source_bucket_incarnation_id) = entry.source_bucket_incarnation_id && source_bucket_incarnation_id != anchor.bucket_incarnation_id { - let previous = entry.state.clone(); - let old_cost = Self::cost_with_state(&entry.intent, &previous); + let old_cost = Self::entry_cost(entry); let detected_at_ms = SystemTime::now() .duration_since(UNIX_EPOCH) .map_or(0, |duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)); @@ -757,10 +750,7 @@ impl PartialWrites { observed_bucket_incarnation_id: anchor.bucket_incarnation_id, detected_at_ms, }; - self.bytes = self - .bytes - .saturating_sub(old_cost) - .saturating_add(Self::cost_with_state(&entry.intent, &entry.state)); + self.bytes = self.bytes.saturating_sub(old_cost).saturating_add(Self::entry_cost(entry)); entry.retry_queued = false; lifecycle_changed = true; continue; @@ -818,11 +808,7 @@ impl PartialWrites { if before == self.entries.len() { return false; } - self.bytes = self - .entries - .values() - .map(|entry| Self::cost_with_state(&entry.intent, &entry.state)) - .sum(); + self.bytes = self.entries.values().map(Self::entry_cost).sum(); self.retry_order.retain(|key| self.entries.contains_key(key)); self.retry_index.retain(|key| self.entries.contains_key(key)); true @@ -1027,6 +1013,18 @@ mod tests { assert!(matches!(previous, ResponsibilityState::HeldUnverifiedLegacy { .. })); assert_eq!(restored.unverified_legacy_count(), 0); assert_eq!(restored.operator_accepted_unverified_count(), 1); + let accepted_entry = restored + .entries + .get(&queue_key(&intent("legacy-lifecycle"))) + .expect("accepted entry"); + assert_eq!( + restored.bytes(), + PartialWrites::cost_with_state_and_audit( + &accepted_entry.intent, + &accepted_entry.state, + accepted_entry.last_operator_acceptance.as_ref(), + ) + ); let (mut items, next_cursor) = restored.list_unverified_legacy(None, 10); let item = items.pop().expect("operator state remains visible"); assert!(next_cursor.is_none()); @@ -1037,6 +1035,19 @@ mod tests { .expect("explicit targeted recheck"); assert!(matches!(accepted, ResponsibilityState::OperatorAcceptedUnverified { .. })); assert_eq!(restored.operator_accepted_unverified_count(), 0); + let active_entry = restored + .entries + .get(&queue_key(&intent("legacy-lifecycle"))) + .expect("active entry"); + assert_eq!( + restored.bytes(), + PartialWrites::cost_with_state_and_audit( + &active_entry.intent, + &active_entry.state, + active_entry.last_operator_acceptance.as_ref(), + ), + "recheck must continue accounting for the retained audit record" + ); restored.mark_persisted(); assert_eq!(restored.ready_keys(Instant::now() + Duration::from_secs(60), 1).len(), 1); } diff --git a/crates/heal/tests/mrf_partial_write_test.rs b/crates/heal/tests/mrf_partial_write_test.rs index aa6e501f4..b68c6d7e7 100644 --- a/crates/heal/tests/mrf_partial_write_test.rs +++ b/crates/heal/tests/mrf_partial_write_test.rs @@ -90,8 +90,26 @@ async fn partial_write_persistence_failure_is_reported_and_retained_for_retry() ); } -#[tokio::test] -async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptance() { +#[test] +fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptance() { + const STACK_SIZE: usize = 8 * 1024 * 1024; + std::thread::Builder::new() + .name("mrf-unbound-lifecycle".to_owned()) + .stack_size(STACK_SIZE) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_current_thread() + .thread_stack_size(STACK_SIZE) + .enable_all() + .build() + .expect("unbound lifecycle runtime should build"); + runtime.block_on(legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptance_inner()); + }) + .expect("unbound lifecycle test thread should spawn") + .join() + .expect("unbound lifecycle test thread should finish"); +} + +async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptance_inner() { use rustfs_common::mrf_channel::{MrfScope, persist_partial_write_intent}; temp_env::async_with_vars([("RUSTFS_HEAL_MRF_ENABLE", Some("true"))], async { diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 6efd75fea..c4e41cc0b 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -1681,19 +1681,21 @@ fn parse_mrf_legacy_responsibility_query(uri: &Uri) -> S3Result<(Option { limit = value .parse::() - .map_err(|_| s3_error!(InvalidArgument, "limit must be an integer between 1 and 256"))?; + .map_err(|_| admin_error(S3ErrorCode::InvalidArgument, "limit must be an integer between 1 and 256"))?; if limit == 0 || limit > MRF_RESPONSIBILITY_LIST_MAX_LIMIT { - return Err(s3_error!(InvalidArgument, "limit must be between 1 and 256")); + return Err(admin_error(S3ErrorCode::InvalidArgument, "limit must be between 1 and 256")); } } - "cursor" | "limit" => return Err(s3_error!(InvalidArgument, "duplicate MRF responsibility query parameter")), - _ => return Err(s3_error!(InvalidArgument, "unknown MRF responsibility query parameter")), + "cursor" | "limit" => { + return Err(admin_error(S3ErrorCode::InvalidArgument, "duplicate MRF responsibility query parameter")); + } + _ => return Err(admin_error(S3ErrorCode::InvalidArgument, "unknown MRF responsibility query parameter")), } } } @@ -1713,14 +1715,19 @@ fn map_mrf_lifecycle_control_error(error: rustfs_heal::heal::mrf_queue::MrfLifec use rustfs_heal::heal::mrf_queue::MrfLifecycleControlError; match error { MrfLifecycleControlError::Unavailable | MrfLifecycleControlError::Persistence => { - s3_error!(ServiceUnavailable, "MRF lifecycle operation could not be completed") + admin_error(S3ErrorCode::ServiceUnavailable, "MRF lifecycle operation could not be completed") } - MrfLifecycleControlError::StaleGeneration => s3_error!(InvalidRequest, "MRF responsibility generation changed"), - MrfLifecycleControlError::NotHeld => s3_error!(InvalidRequest, "MRF responsibility is not held as unverified legacy"), - MrfLifecycleControlError::IncarnationChanged => { - s3_error!(InvalidRequest, "bucket incarnation changed; refresh the MRF responsibility listing") + MrfLifecycleControlError::StaleGeneration => { + admin_error(S3ErrorCode::InvalidRequest, "MRF responsibility generation changed") } - MrfLifecycleControlError::InvalidAction(reason) => s3_error!(InvalidRequest, "{reason}"), + MrfLifecycleControlError::NotHeld => { + admin_error(S3ErrorCode::InvalidRequest, "MRF responsibility is not held as unverified legacy") + } + MrfLifecycleControlError::IncarnationChanged => admin_error( + S3ErrorCode::InvalidRequest, + "bucket incarnation changed; refresh the MRF responsibility listing", + ), + MrfLifecycleControlError::InvalidAction(reason) => admin_error(S3ErrorCode::InvalidRequest, reason), } } @@ -1736,10 +1743,10 @@ impl Operation for MrfLegacyResponsibilitiesHandler { rustfs_heal::heal::mrf_queue::list_legacy_responsibilities(cursor, limit), ) .await - .map_err(|_| s3_error!(ServiceUnavailable, "MRF responsibility listing timed out"))? + .map_err(|_| admin_error(S3ErrorCode::ServiceUnavailable, "MRF responsibility listing timed out"))? .map_err(map_mrf_lifecycle_control_error)?; - let body = - serde_json::to_vec(&snapshot).map_err(|_| s3_error!(InternalError, "failed to encode MRF responsibility listing"))?; + let body = serde_json::to_vec(&snapshot) + .map_err(|_| admin_error(S3ErrorCode::InternalError, "failed to encode MRF responsibility listing"))?; Ok(json_response(StatusCode::OK, body)) } } @@ -1754,10 +1761,10 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { .input .store_all_limited(rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE) .await - .map_err(|_| s3_error!(InvalidRequest, "MRF responsibility action body is too large or unreadable"))?; - let action: MrfLegacyResponsibilityActionRequest = - serde_json::from_slice(&bytes).map_err(|_| s3_error!(InvalidRequest, "invalid MRF responsibility action body"))?; - validate_mrf_legacy_responsibility_action(&action).map_err(|reason| s3_error!(InvalidRequest, "{reason}"))?; + .map_err(|_| admin_error(S3ErrorCode::InvalidRequest, "MRF responsibility action body is too large or unreadable"))?; + let action: MrfLegacyResponsibilityActionRequest = serde_json::from_slice(&bytes) + .map_err(|_| admin_error(S3ErrorCode::InvalidRequest, "invalid MRF responsibility action body"))?; + validate_mrf_legacy_responsibility_action(&action).map_err(|reason| admin_error(S3ErrorCode::InvalidRequest, reason))?; let (state, data_verified, durable_responsibility_retained) = match action.action { MrfLegacyResponsibilityAction::Refresh => { timeout( @@ -1768,7 +1775,7 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { ), ) .await - .map_err(|_| s3_error!(ServiceUnavailable, "MRF responsibility refresh timed out"))? + .map_err(|_| admin_error(S3ErrorCode::ServiceUnavailable, "MRF responsibility refresh timed out"))? .map_err(map_mrf_lifecycle_control_error)?; ("refreshed", false, true) } @@ -1781,18 +1788,20 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { ), ) .await - .map_err(|_| s3_error!(ServiceUnavailable, "MRF responsibility recheck timed out"))? + .map_err(|_| admin_error(S3ErrorCode::ServiceUnavailable, "MRF responsibility recheck timed out"))? .map_err(map_mrf_lifecycle_control_error)?; ("active", false, true) } MrfLegacyResponsibilityAction::AcceptUnverifiedRisk => { - let reason = action.reason.ok_or_else(|| s3_error!(InvalidRequest, "reason is required"))?; + let reason = action + .reason + .ok_or_else(|| admin_error(S3ErrorCode::InvalidRequest, "reason is required"))?; let reference = action .reference - .ok_or_else(|| s3_error!(InvalidRequest, "reference is required"))?; + .ok_or_else(|| admin_error(S3ErrorCode::InvalidRequest, "reference is required"))?; let request_id = action .request_id - .ok_or_else(|| s3_error!(InvalidRequest, "requestId is required"))?; + .ok_or_else(|| admin_error(S3ErrorCode::InvalidRequest, "requestId is required"))?; timeout( Duration::from_secs(30), rustfs_heal::heal::mrf_queue::accept_unverified_legacy_risk( @@ -1807,7 +1816,12 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { ), ) .await - .map_err(|_| s3_error!(ServiceUnavailable, "MRF risk disposition timed out; read status before retrying"))? + .map_err(|_| { + admin_error( + S3ErrorCode::ServiceUnavailable, + "MRF risk disposition timed out; read status before retrying", + ) + })? .map_err(map_mrf_lifecycle_control_error)?; ("operatorAcceptedUnverified", false, true) } @@ -1818,7 +1832,7 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { data_verified, durable_responsibility_retained, }) - .map_err(|_| s3_error!(InternalError, "failed to encode MRF responsibility action response"))?; + .map_err(|_| admin_error(S3ErrorCode::InternalError, "failed to encode MRF responsibility action response"))?; Ok(json_response(StatusCode::OK, body)) } }