gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-12 23:52:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(heal): group MRF risk acceptance inputs

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-09-30 08:30:58 +08:00
parent cf9c1076b1
commit a3c2070eac
4 changed files with 116 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/mrf_queue.rs
+25 -87
View File
@@ -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<Result<(), MrfLifecycleControlError>>,
},
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<Result<(), MrfLifecycleControlError>>,
},
}
@@ -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");
crates/heal/src/heal/mrf_queue/partial_write.rs
+63 -66
View File
@@ -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<MrfOperatorAcceptance>), &'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 { .. }));
crates/heal/tests/mrf_partial_write_test.rs
+18 -18
View File
@@ -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");
rustfs/src/admin/handlers/heal.rs
+10 -8
View File
@@ -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