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

fix(admin): route MRF errors through gateway facade

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:04:13 +08:00
parent df0d832c43
commit 0c3617f3f8
4 changed files with 120 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/mrf_queue.rs
+3 -1
View File
@@ -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::<usize>(),
crates/heal/src/heal/mrf_queue/partial_write.rs
+58 -47
View File
@@ -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);
}
crates/heal/tests/mrf_partial_write_test.rs
+20 -2
View File
@@ -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 {
rustfs/src/admin/handlers/heal.rs
+39 -25
View File
@@ -1681,19 +1681,21 @@ fn parse_mrf_legacy_responsibility_query(uri: &Uri) -> S3Result<(Option<uuid::Uu
cursor = Some(
value
.parse()
.map_err(|_| s3_error!(InvalidArgument, "cursor must be a UUID"))?,
.map_err(|_| admin_error(S3ErrorCode::InvalidArgument, "cursor must be a UUID"))?,
);
}
"limit" if seen.insert("limit") => {
limit = value
.parse::<usize>()
.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))
}
}