From a3c2070eac5609689df87e6a567dca795b528aca Mon Sep 17 00:00:00 2001 From: Hauser Date: Wed, 30 Sep 2026 08:30:58 +0800 Subject: [PATCH] refactor(heal): group MRF risk acceptance inputs Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- crates/heal/src/heal/mrf_queue.rs | 112 ++++----------- .../heal/src/heal/mrf_queue/partial_write.rs | 129 +++++++++--------- crates/heal/tests/mrf_partial_write_test.rs | 36 ++--- rustfs/src/admin/handlers/heal.rs | 18 +-- 4 files changed, 116 insertions(+), 179 deletions(-) diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs index 77bdba8db..4014fc33e 100644 --- a/crates/heal/src/heal/mrf_queue.rs +++ b/crates/heal/src/heal/mrf_queue.rs @@ -68,6 +68,18 @@ pub enum MrfLifecycleControlError { Persistence, } +#[derive(Debug)] +pub struct MrfLegacyRiskAcceptanceRequest { + pub responsibility_id: Uuid, + pub expected_bucket_incarnation_id: Uuid, + pub acknowledge_unknown_source_incarnation: bool, + pub acknowledge_incarnation_mismatch: bool, + pub actor: String, + pub reason: String, + pub reference: String, + pub request_id: Uuid, +} + #[derive(Clone, Debug, Serialize)] #[serde(rename_all = "camelCase")] pub struct MrfLegacyResponsibilitySnapshot { @@ -142,32 +154,13 @@ pub async fn refresh_legacy_responsibility( receiver.await.map_err(|_| MrfLifecycleControlError::Unavailable)? } -pub async fn accept_unverified_legacy_risk( - responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, - acknowledge_unknown_source_incarnation: bool, - acknowledge_incarnation_mismatch: bool, - actor: String, - reason: String, - reference: String, - request_id: Uuid, -) -> Result<(), MrfLifecycleControlError> { +pub async fn accept_unverified_legacy_risk(request: MrfLegacyRiskAcceptanceRequest) -> Result<(), MrfLifecycleControlError> { let sender = GLOBAL_MRF_LIFECYCLE_SENDER .get() .ok_or(MrfLifecycleControlError::Unavailable)?; let (response, receiver) = tokio::sync::oneshot::channel(); sender - .send(MrfLifecycleCommand::AcceptUnverifiedRisk { - responsibility_id, - expected_bucket_incarnation_id, - acknowledge_unknown_source_incarnation, - acknowledge_incarnation_mismatch, - actor, - reason, - reference, - request_id, - response, - }) + .send(MrfLifecycleCommand::AcceptUnverifiedRisk { request, response }) .await .map_err(|_| MrfLifecycleControlError::Unavailable)?; receiver.await.map_err(|_| MrfLifecycleControlError::Unavailable)? @@ -190,14 +183,7 @@ enum MrfLifecycleCommand { response: tokio::sync::oneshot::Sender>, }, AcceptUnverifiedRisk { - responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, - acknowledge_unknown_source_incarnation: bool, - acknowledge_incarnation_mismatch: bool, - actor: String, - reason: String, - reference: String, - request_id: Uuid, + request: MrfLegacyRiskAcceptanceRequest, response: tokio::sync::oneshot::Sender>, }, } @@ -1107,58 +1093,32 @@ impl MrfRuntime { async fn accept_legacy_risk( &mut self, manager: &HealManager, - responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, - acknowledge_unknown_source_incarnation: bool, - acknowledge_incarnation_mismatch: bool, - actor: String, - reason: String, - reference: String, - request_id: Uuid, + request: MrfLegacyRiskAcceptanceRequest, ) -> Result<(), MrfLifecycleControlError> { + let responsibility_id = request.responsibility_id; if self .partial_writes - .intent_for_responsibility(responsibility_id) + .intent_for_responsibility(request.responsibility_id) .is_some_and(|(_, state, _)| { matches!( state, partial_write::ResponsibilityState::OperatorAcceptedUnverified { request_id: stored_request_id, .. - } if stored_request_id == request_id + } if stored_request_id == request.request_id ) }) { self.partial_writes - .record_operator_acceptance( - self.queue.byte_budget.saturating_sub(self.queue.bytes()), - responsibility_id, - expected_bucket_incarnation_id, - acknowledge_unknown_source_incarnation, - acknowledge_incarnation_mismatch, - actor, - reason, - reference, - request_id, - ) + .record_operator_acceptance(self.queue.byte_budget.saturating_sub(self.queue.bytes()), request) .map_err(map_mrf_lifecycle_state_error)?; return Ok(()); } - self.verify_legacy_generation_incarnation(manager, responsibility_id, expected_bucket_incarnation_id) + self.verify_legacy_generation_incarnation(manager, request.responsibility_id, request.expected_bucket_incarnation_id) .await?; let (previous, previous_acceptance) = self .partial_writes - .record_operator_acceptance( - self.queue.byte_budget.saturating_sub(self.queue.bytes()), - responsibility_id, - expected_bucket_incarnation_id, - acknowledge_unknown_source_incarnation, - acknowledge_incarnation_mismatch, - actor, - reason, - reference, - request_id, - ) + .record_operator_acceptance(self.queue.byte_budget.saturating_sub(self.queue.bytes()), request) .map_err(map_mrf_lifecycle_state_error)?; self.dirty = true; if !self.flush().await { @@ -1903,30 +1863,8 @@ async fn run_mrf_consumer( .await; let _ = response.send(result); } - MrfLifecycleCommand::AcceptUnverifiedRisk { - responsibility_id, - expected_bucket_incarnation_id, - acknowledge_unknown_source_incarnation, - acknowledge_incarnation_mismatch, - actor, - reason, - reference, - request_id, - response, - } => { - let result = runtime - .accept_legacy_risk( - manager.as_ref(), - responsibility_id, - expected_bucket_incarnation_id, - acknowledge_unknown_source_incarnation, - acknowledge_incarnation_mismatch, - actor, - reason, - reference, - request_id, - ) - .await; + MrfLifecycleCommand::AcceptUnverifiedRisk { request, response } => { + let result = runtime.accept_legacy_risk(manager.as_ref(), request).await; let _ = response.send(result); } } @@ -2162,7 +2100,7 @@ mod tests { request_id: accepted_request_id, }, }; - let records = vec![record.clone(), accepted.clone()]; + let records = vec![record, accepted]; let encoded = encode_mrf_lifecycle_checkpoint(owner, 42, records.clone(), 8192) .expect("lifecycle checkpoint should fit the configured bound"); let decoded = decode_mrf_lifecycle_checkpoint(&encoded, 8192).expect("valid lifecycle checkpoint"); diff --git a/crates/heal/src/heal/mrf_queue/partial_write.rs b/crates/heal/src/heal/mrf_queue/partial_write.rs index 11fa699d8..bcecd0dd9 100644 --- a/crates/heal/src/heal/mrf_queue/partial_write.rs +++ b/crates/heal/src/heal/mrf_queue/partial_write.rs @@ -12,7 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::{HealManager, MrfConsumerConfig, MrfDurableRepairAnchor, MrfIntent, MrfQueueKey, queue_key, submit_mrf_heal_request}; +use super::{ + HealManager, MrfConsumerConfig, MrfDurableRepairAnchor, MrfIntent, MrfLegacyRiskAcceptanceRequest, MrfQueueKey, queue_key, + submit_mrf_heal_request, +}; use rustfs_common::mrf_channel::{MrfDurableAdmissionError, MrfIngressResult, release_mrf_intent, try_rearm_mrf_replay_intent}; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; @@ -421,15 +424,18 @@ impl PartialWrites { pub(super) fn record_operator_acceptance( &mut self, byte_budget: usize, - responsibility_id: Uuid, - expected_bucket_incarnation_id: Uuid, - acknowledge_unknown_source_incarnation: bool, - acknowledge_incarnation_mismatch: bool, - actor: String, - reason: String, - reference: String, - request_id: Uuid, + request: MrfLegacyRiskAcceptanceRequest, ) -> Result<(ResponsibilityState, Option), &'static str> { + let MrfLegacyRiskAcceptanceRequest { + responsibility_id, + expected_bucket_incarnation_id, + acknowledge_unknown_source_incarnation, + acknowledge_incarnation_mismatch, + actor, + reason, + reference, + request_id, + } = request; validate_operator_audit_fields(&actor, &reason, &reference, request_id)?; let entry = self .entries @@ -866,6 +872,7 @@ fn unix_now_ms() -> u64 { #[cfg(test)] mod tests { use super::*; + use crate::heal::mrf_queue::MrfLegacyRiskAcceptanceRequest; use rustfs_common::mrf_channel::{MrfKind, MrfScope}; use std::sync::Arc; use std::time::Duration; @@ -890,6 +897,26 @@ mod tests { intent } + fn risk_acceptance( + responsibility_id: Uuid, + expected_bucket_incarnation_id: Uuid, + acknowledge_unknown_source_incarnation: bool, + acknowledge_incarnation_mismatch: bool, + reason: &str, + reference: &str, + ) -> MrfLegacyRiskAcceptanceRequest { + MrfLegacyRiskAcceptanceRequest { + responsibility_id, + expected_bucket_incarnation_id, + acknowledge_unknown_source_incarnation, + acknowledge_incarnation_mismatch, + actor: "operator-a".to_string(), + reason: reason.to_string(), + reference: reference.to_string(), + request_id: Uuid::new_v4(), + } + } + #[test] fn partial_write_retention_bounds_count_and_bytes_without_evicting_responsibility() { let first = intent("a"); @@ -1000,14 +1027,14 @@ mod tests { let (previous, _) = restored .record_operator_acceptance( 8192, - responsibility_id, - incarnation, - true, - false, - "operator-a".to_string(), - "Reviewed against trusted backup; accepting unresolved identity risk".to_string(), - "INC-1234".to_string(), - Uuid::new_v4(), + risk_acceptance( + responsibility_id, + incarnation, + true, + false, + "Reviewed against trusted backup; accepting unresolved identity risk", + "INC-1234", + ), ) .expect("explicit risk acceptance"); assert!(matches!(previous, ResponsibilityState::HeldUnverifiedLegacy { .. })); @@ -1111,47 +1138,17 @@ mod tests { assert!( writes - .record_operator_acceptance( - 8192, - id, - Uuid::new_v4(), - true, - false, - "operator-a".to_string(), - "reason".to_string(), - "INC-1234".to_string(), - Uuid::new_v4(), - ) + .record_operator_acceptance(8192, risk_acceptance(id, Uuid::new_v4(), true, false, "reason", "INC-1234")) .is_err() ); assert!( writes - .record_operator_acceptance( - 8192, - Uuid::new_v4(), - incarnation, - true, - false, - "operator-a".to_string(), - "reason".to_string(), - "INC-1234".to_string(), - Uuid::new_v4(), - ) + .record_operator_acceptance(8192, risk_acceptance(Uuid::new_v4(), incarnation, true, false, "reason", "INC-1234")) .is_err() ); assert!( writes - .record_operator_acceptance( - 8192, - id, - incarnation, - true, - false, - "operator-a".to_string(), - " ".to_string(), - "INC-1234".to_string(), - Uuid::new_v4(), - ) + .record_operator_acceptance(8192, risk_acceptance(id, incarnation, true, false, " ", "INC-1234")) .is_err() ); assert_eq!(writes.unverified_legacy_count(), 1); @@ -1179,28 +1176,28 @@ mod tests { writes .record_operator_acceptance( 8192, - responsibility_id, - observed, - false, - false, - "operator-a".to_string(), - "The old bucket generation is no longer available".to_string(), - "INC-5678".to_string(), - Uuid::new_v4(), + risk_acceptance( + responsibility_id, + observed, + false, + false, + "The old bucket generation is no longer available", + "INC-5678", + ), ) .is_err() ); let (accepted, _) = writes .record_operator_acceptance( 8192, - responsibility_id, - observed, - false, - true, - "operator-a".to_string(), - "The old bucket generation is no longer available".to_string(), - "INC-5678".to_string(), - Uuid::new_v4(), + risk_acceptance( + responsibility_id, + observed, + false, + true, + "The old bucket generation is no longer available", + "INC-5678", + ), ) .expect("generation mismatch requires and accepts explicit acknowledgment"); assert!(matches!(accepted, ResponsibilityState::BucketIncarnationChanged { .. })); diff --git a/crates/heal/tests/mrf_partial_write_test.rs b/crates/heal/tests/mrf_partial_write_test.rs index b68c6d7e7..2a079be65 100644 --- a/crates/heal/tests/mrf_partial_write_test.rs +++ b/crates/heal/tests/mrf_partial_write_test.rs @@ -167,16 +167,16 @@ async fn legacy_unbound_generation_is_parked_and_requires_explicit_risk_acceptan mrf_queue::recheck_legacy_responsibility(entry.responsibility_id, entry.bucket_incarnation_id).await, Err(rustfs_heal::heal::mrf_queue::MrfLifecycleControlError::InvalidAction(_)) )); - mrf_queue::accept_unverified_legacy_risk( - entry.responsibility_id, - entry.bucket_incarnation_id, - true, - false, - "integration-test-operator".to_string(), - "The old journal has no source bucket-generation binding".to_string(), - "TEST-ISSUE-2682-UNBOUND".to_string(), - uuid::Uuid::new_v4(), - ) + mrf_queue::accept_unverified_legacy_risk(mrf_queue::MrfLegacyRiskAcceptanceRequest { + responsibility_id: entry.responsibility_id, + expected_bucket_incarnation_id: entry.bucket_incarnation_id, + acknowledge_unknown_source_incarnation: true, + acknowledge_incarnation_mismatch: false, + actor: "integration-test-operator".to_string(), + reason: "The old journal has no source bucket-generation binding".to_string(), + reference: "TEST-ISSUE-2682-UNBOUND".to_string(), + request_id: uuid::Uuid::new_v4(), + }) .await .expect("unbound risk disposition requires explicit acknowledgment and must be durable"); assert!( @@ -752,16 +752,16 @@ async fn partial_write_sigkill_replay_scenario(protected: bool) { let responsibility_id = held.responsibility_id; let bucket_incarnation_id = held.bucket_incarnation_id; let request_id = uuid::Uuid::new_v4(); - mrf_queue::accept_unverified_legacy_risk( + mrf_queue::accept_unverified_legacy_risk(mrf_queue::MrfLegacyRiskAcceptanceRequest { responsibility_id, - bucket_incarnation_id, - true, - false, - "integration-test-operator".to_string(), - "The operator has accepted that legacy object identity cannot be proven automatically".to_string(), - "TEST-ISSUE-2682".to_string(), + expected_bucket_incarnation_id: bucket_incarnation_id, + acknowledge_unknown_source_incarnation: true, + acknowledge_incarnation_mismatch: false, + actor: "integration-test-operator".to_string(), + reason: "The operator has accepted that legacy object identity cannot be proven automatically".to_string(), + reference: "TEST-ISSUE-2682".to_string(), request_id, - ) + }) .await .expect("explicit risk acceptance must persist before success"); assert!(snapshot_contains("crash.bin").await, "risk acceptance must retain the MRF responsibility"); diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index c4e41cc0b..f6c02aa0c 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -1805,14 +1805,16 @@ impl Operation for MrfLegacyResponsibilitiesActionHandler { timeout( Duration::from_secs(30), rustfs_heal::heal::mrf_queue::accept_unverified_legacy_risk( - action.responsibility_id, - action.expected_bucket_incarnation_id, - action.acknowledge_unknown_source_incarnation == Some(true), - action.acknowledge_bucket_incarnation_mismatch == Some(true), - actor, - reason, - reference, - request_id, + rustfs_heal::heal::mrf_queue::MrfLegacyRiskAcceptanceRequest { + responsibility_id: action.responsibility_id, + expected_bucket_incarnation_id: action.expected_bucket_incarnation_id, + acknowledge_unknown_source_incarnation: action.acknowledge_unknown_source_incarnation == Some(true), + acknowledge_incarnation_mismatch: action.acknowledge_bucket_incarnation_mismatch == Some(true), + actor, + reason, + reference, + request_id, + }, ), ) .await