gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 16:37:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(rpc): dual-write a typed not-initialized code on control-plane responses (#6684)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-27 03:15:32 +08:00
committed by GitHub
parent f4cc919401
commit a42046b79c
9 changed files with 383 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cluster/rpc/peer_rest_client.rs
+161 -84
View File
@@ -113,6 +113,21 @@ fn peer_failure_without_details(op: &str, bucket: Option<&str>) -> Error {
}
}
/// Decode a control-plane response failure. Peers at or above the typed
/// `ControlPlaneErrorCode` change (backlog#1845) carry a machine-readable
/// discriminant beside the legacy `error_info` string; prefer it, then fall
/// back to the string, then to the detail-free per-op failure.
/// RUSTFS_COMPAT_TODO(not-initialized-error-code-v1): string fallback for peers that predate the typed wire code. Remove after the minimum supported RustFS peer version always sends error_code.
fn control_plane_failure(op: &str, bucket: Option<&str>, error_code: Option<i32>, error_info: Option<String>) -> Error {
if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32) {
return Error::RemoteNotInitialized;
}
match error_info {
Some(msg) => Error::other(msg),
None => peer_failure_without_details(op, bucket),
}
}
fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result<BucketStats> {
if !response.success {
return Err(Error::other(
@@ -845,10 +860,12 @@ impl PeerRestClient {
let response = client.local_storage_info(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("local_storage_info", None));
return Err(control_plane_failure(
"local_storage_info",
None,
response.error_code,
response.error_info,
));
}
let data = response.storage_info;
@@ -1235,11 +1252,10 @@ impl PeerRestClient {
Err(status) => return Err(status.into()),
};
if !response.success {
return Err(Error::other(
response
.error_info
.unwrap_or_else(|| "peer background heal status failed without an error".to_string()),
));
return Err(match (response.error_code, response.error_info) {
(None, None) => Error::other("peer background heal status failed without an error"),
(error_code, error_info) => control_plane_failure("background_heal_status", None, error_code, error_info),
});
}
Ok(Some(response.bg_heal_state.to_vec()))
}
@@ -1267,11 +1283,12 @@ impl PeerRestClient {
Err(status) => return Err(status.into()),
};
if !response.success {
return Err(Error::other(
response
.error_info
.unwrap_or_else(|| "peer replacement recovery status failed without an error".to_string()),
));
return Err(match (response.error_code, response.error_info) {
(None, None) => Error::other("peer replacement recovery status failed without an error"),
(error_code, error_info) => {
control_plane_failure("replacement_recovery_status", None, error_code, error_info)
}
});
}
Ok(Some(response.recovery_status.to_vec()))
}
@@ -1489,10 +1506,12 @@ impl PeerRestClient {
let response = client.load_bucket_metadata(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
return Err(control_plane_failure(
"load_bucket_metadata",
Some(bucket),
response.error_code,
response.error_info,
));
}
Ok(())
}
@@ -1531,10 +1550,7 @@ impl PeerRestClient {
let response = client.delete_policy(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("delete_policy", None));
return Err(control_plane_failure("delete_policy", None, response.error_code, response.error_info));
}
Ok(())
}
@@ -1554,10 +1570,7 @@ impl PeerRestClient {
let response = client.load_policy(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_policy", None));
return Err(control_plane_failure("load_policy", None, response.error_code, response.error_info));
}
Ok(())
}
@@ -1579,10 +1592,12 @@ impl PeerRestClient {
let response = client.load_policy_mapping(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_policy_mapping", None));
return Err(control_plane_failure(
"load_policy_mapping",
None,
response.error_code,
response.error_info,
));
}
Ok(())
}
@@ -1602,10 +1617,7 @@ impl PeerRestClient {
let response = client.delete_user(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("delete_user", None));
return Err(control_plane_failure("delete_user", None, response.error_code, response.error_info));
}
Ok(())
}
@@ -1625,10 +1637,12 @@ impl PeerRestClient {
let response = client.delete_service_account(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("delete_service_account", None));
return Err(control_plane_failure(
"delete_service_account",
None,
response.error_code,
response.error_info,
));
}
Ok(())
}
@@ -1649,10 +1663,7 @@ impl PeerRestClient {
let response = client.load_user(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_user", None));
return Err(control_plane_failure("load_user", None, response.error_code, response.error_info));
}
Ok(())
}
@@ -1672,10 +1683,12 @@ impl PeerRestClient {
let response = client.load_service_account(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_service_account", None));
return Err(control_plane_failure(
"load_service_account",
None,
response.error_code,
response.error_info,
));
}
Ok(())
}
@@ -1695,10 +1708,7 @@ impl PeerRestClient {
let response = client.load_group(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_group", None));
return Err(control_plane_failure("load_group", None, response.error_code, response.error_info));
}
Ok(())
}
@@ -1716,10 +1726,12 @@ impl PeerRestClient {
let response = client.reload_site_replication_config(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("reload_site_replication_config", None));
return Err(control_plane_failure(
"reload_site_replication_config",
None,
response.error_code,
response.error_info,
));
}
Ok(())
}
@@ -1987,10 +1999,7 @@ impl PeerRestClient {
let response = client.reload_pool_meta(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("reload_pool_meta", None));
return Err(control_plane_failure("reload_pool_meta", None, response.error_code, response.error_info));
}
Ok(())
@@ -2011,10 +2020,7 @@ impl PeerRestClient {
let response = client.stop_rebalance(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("stop_rebalance", None));
return Err(control_plane_failure("stop_rebalance", None, response.error_code, response.error_info));
}
Ok(())
@@ -2045,10 +2051,12 @@ impl PeerRestClient {
"peer rebalance metadata response"
);
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_rebalance_meta", None));
return Err(control_plane_failure(
"load_rebalance_meta",
None,
response.error_code,
response.error_info,
));
}
Ok(())
@@ -2073,10 +2081,12 @@ impl PeerRestClient {
let response = client.start_decommission(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("start_decommission", None));
return Err(control_plane_failure(
"start_decommission",
None,
response.error_code,
response.error_info,
));
}
Ok(())
@@ -2097,10 +2107,12 @@ impl PeerRestClient {
let response = client.cancel_decommission(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("decommission_cancel", None));
return Err(control_plane_failure(
"decommission_cancel",
None,
response.error_code,
response.error_info,
));
}
Ok(())
@@ -2121,10 +2133,12 @@ impl PeerRestClient {
let response = client.clear_decommission(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("clear_decommission", None));
return Err(control_plane_failure(
"clear_decommission",
None,
response.error_code,
response.error_info,
));
}
Ok(())
@@ -2180,7 +2194,7 @@ impl PeerRestClient {
Err(status) => return tier_config_reload_status_outcome(status),
};
if !response.success {
return tier_config_reload_remote_failure(response.error_info);
return tier_config_reload_remote_failure(response.error_code, response.error_info);
}
TierConfigReloadOutcome::Success
@@ -2239,7 +2253,13 @@ fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
/// reload every `TIER_CONFIG_RELOAD_RETRY_CAP`, and `Terminal` stays reachable
/// for transport and gRPC status failures, which is where a genuinely
/// unrecoverable peer surfaces.
fn tier_config_reload_remote_failure(error_info: Option<String>) -> TierConfigReloadOutcome {
fn tier_config_reload_remote_failure(error_code: Option<i32>, error_info: Option<String>) -> TierConfigReloadOutcome {
// Remote rejections are transient by design (see the doc comment above);
// the typed not-initialized code keeps the error typed for downstream
// classifiers instead of a bare string (backlog#1845).
if error_code == Some(rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32) {
return TierConfigReloadOutcome::TransientRetrySameChannel(Error::RemoteNotInitialized);
}
TierConfigReloadOutcome::TransientRetrySameChannel(Error::other(error_info.unwrap_or_default()))
}
@@ -2278,6 +2298,63 @@ mod tests {
use temp_env::async_with_vars;
use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
#[test]
fn control_plane_failure_prefers_typed_not_initialized_code() {
use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
let code = Some(ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32);
// Typed code wins even when the legacy string is present (dual-write).
let err = control_plane_failure("load_bucket_metadata", Some("b"), code, Some("errServerNotInitialized".to_string()));
assert!(matches!(err, Error::RemoteNotInitialized));
assert!(crate::error::is_err_not_initialized(&err), "typed variant must satisfy the predicate");
// Legacy peers: no code, string only — the substring fallback still classifies.
let err = control_plane_failure("load_bucket_metadata", Some("b"), None, Some("errServerNotInitialized".to_string()));
assert!(crate::error::is_err_not_initialized(&err), "legacy string form must keep classifying");
// No code, no string: detail-free per-op failure, not misread as not-initialized.
let err = control_plane_failure("load_bucket_metadata", Some("b"), None, None);
assert!(!crate::error::is_err_not_initialized(&err));
assert!(err.to_string().contains("load_bucket_metadata"));
// Unspecified code behaves like no code.
let err = control_plane_failure(
"load_bucket_metadata",
None,
Some(ControlPlaneErrorCode::ControlPlaneErrorUnspecified as i32),
Some("boom".to_string()),
);
assert!(!matches!(err, Error::RemoteNotInitialized));
assert_eq!(err.to_string(), "Io error: boom");
}
#[test]
fn control_plane_not_initialized_wire_value_is_pinned() {
// The discriminant is wire contract: old peers ignore it, but a renumber
// would silently flip classification on mixed-version clusters.
use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorUnspecified as i32, 0);
assert_eq!(ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32, 1);
}
#[test]
fn tier_config_reload_remote_failure_keeps_typed_not_initialized() {
use rustfs_protos::proto_gen::node_service::ControlPlaneErrorCode;
let outcome = tier_config_reload_remote_failure(
Some(ControlPlaneErrorCode::ControlPlaneErrorNotInitialized as i32),
Some("errServerNotInitialized".to_string()),
);
match outcome {
TierConfigReloadOutcome::TransientRetrySameChannel(err) => {
assert!(matches!(err, Error::RemoteNotInitialized));
}
TierConfigReloadOutcome::TransientReconnect(err) | TierConfigReloadOutcome::Terminal(err) => {
panic!("not-initialized must stay retry-same-channel, got {err}")
}
TierConfigReloadOutcome::Success => panic!("a rejection cannot classify as success"),
}
}
#[test]
fn scanner_publication_lease_response_rejects_stale_generation_and_session() {
let token = Uuid::new_v4();
@@ -3031,11 +3108,11 @@ mod tests {
// retired: the channel is healthy, so the rejection reflects remote state
// that the next attempt can find healed.
assert!(matches!(
tier_config_reload_remote_failure(Some("backend unavailable".to_string())),
tier_config_reload_remote_failure(None, Some("backend unavailable".to_string())),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
));
assert!(matches!(
tier_config_reload_remote_failure(Some("errServerNotInitialized".to_string())),
tier_config_reload_remote_failure(None, Some("errServerNotInitialized".to_string())),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
));
assert!(matches!(
@@ -3080,7 +3157,7 @@ mod tests {
] {
assert!(
matches!(
tier_config_reload_remote_failure(Some(error_info.to_string())),
tier_config_reload_remote_failure(None, Some(error_info.to_string())),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
),
"a peer that rejected the apply must stay retryable so it converges: {error_info}"
@@ -3089,7 +3166,7 @@ mod tests {
// An absent error message is still a rejection, not a reason to stop.
assert!(matches!(
tier_config_reload_remote_failure(None),
tier_config_reload_remote_failure(None, None),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
));
crates/ecstore/src/error/mod.rs
+18 -1
View File
@@ -228,6 +228,14 @@ pub enum StorageError {
/// during quorum aggregation (backlog#1845).
#[error("remote rpc client unavailable: {0}")]
RemoteClientUnavailable(String),
/// A peer answered a control-plane RPC but its storage/IAM layer is not
/// initialized yet. Typed form of the legacy "errServerNotInitialized"
/// error_info string (backlog#1845); the wire carries it as
/// `ControlPlaneErrorCode::ControlPlaneErrorNotInitialized` alongside the
/// legacy string for rolling-upgrade compatibility.
#[error("remote peer not initialized")]
RemoteNotInitialized,
}
impl From<crate::erasure::coding::ErasureConstructionError> for StorageError {
@@ -581,6 +589,7 @@ impl Clone for StorageError {
limit: *limit,
},
StorageError::RemoteClientUnavailable(detail) => StorageError::RemoteClientUnavailable(detail.clone()),
StorageError::RemoteNotInitialized => StorageError::RemoteNotInitialized,
}
}
}
@@ -670,6 +679,7 @@ impl StorageError {
StorageError::NamespaceLockQuorumUnavailable { .. } => StorageErrorCode::NamespaceLockQuorumUnavailable,
StorageError::QuotaExceeded { .. } => StorageErrorCode::QuotaExceeded,
StorageError::RemoteClientUnavailable(_) => StorageErrorCode::RemoteClientUnavailable,
StorageError::RemoteNotInitialized => StorageErrorCode::RemoteNotInitialized,
}
}
@@ -800,6 +810,7 @@ impl StorageError {
limit: Default::default(),
}),
StorageErrorCode::RemoteClientUnavailable => Some(StorageError::RemoteClientUnavailable(Default::default())),
StorageErrorCode::RemoteNotInitialized => Some(StorageError::RemoteNotInitialized),
}
}
}
@@ -943,7 +954,13 @@ pub fn is_err_operation_canceled(err: &Error) -> bool {
#[allow(dead_code, reason = "predicate asserted by this file's tests (backlog#1823)")]
pub fn is_err_not_initialized(err: &Error) -> bool {
err.to_string().contains("errServerNotInitialized") || err.to_string().contains("ServerNotInitialized")
// Typed-first: peers at or above the ControlPlaneErrorCode change decode to
// the typed variant. The substring form only matches legacy peers' string
// responses.
// RUSTFS_COMPAT_TODO(not-initialized-error-code-v1): substring fallback for peers that predate the typed wire code. Remove after the minimum supported RustFS peer version always sends error_code.
matches!(err, StorageError::RemoteNotInitialized)
|| err.to_string().contains("errServerNotInitialized")
|| err.to_string().contains("ServerNotInitialized")
}
/// Strict "not found" predicate that only matches genuine object/version/volume