gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-30 08:49:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ecstore): type internode client-acquisition failures for quorum buckets (#6619)

Browse Source 
* feat(ecstore): type internode client-acquisition failures for stable quorum buckets

Backlog#1845 step 3, first typed family. The largest other(format!) message family in ecstore was 'can not get client, err: {detail}' (~50 production sites): every internode RPC that fails to acquire a client wrapped the dial/auth error with per-peer detail into DiskError::other / StorageError::other, whose Io equality compares the rendered message. N disks failing for this same cause therefore counted as N distinct errors in reduce_errs, starving quorum aggregation, and remote_disk call sites double-wrapped the message on top of get_client's own wrap.

Introduce DiskError::RemoteClientUnavailable(String) (wire code 0x2B) and its StorageError twin (StorageErrorCode 0x54): equality and hashing use the wire code alone, so same-cause failures land in one quorum bucket regardless of per-peer detail, while Display keeps the detail so substring classifiers (network needles, heal recoverability) keep reading it unchanged. Wire encoding carries the rendered detail in error_info and decode restores the typed variant; old peers fall back to the legacy string form gracefully.

Call sites: remote_disk get_client/get_bulk_client/offline-bypass/recovery-probe now construct the typed variant and the ~60 redundant double-wrap map_errs are gone; peer_rest_client's three client getters and offline gates, peer_s3_client, and admin_server_info follow. The tier-config-reload connection classifier's anchored 'can not get client' substring check becomes a typed match on the variant (the string form is retired and now classifies as Terminal, pinned by test).

Ref rustfs/backlog#1845

* chore(ci): refresh error other ratchet baseline

* fix(ecstore): classify typed client network failures
This commit is contained in:
Zhengchao An
2026-08-26 11:24:56 +08:00
committed by GitHub
parent 7cac528de3
commit c0c208d89a
9 changed files with 176 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cluster/rpc/client.rs
+11
View File
@@ -198,6 +198,7 @@ pub(crate) fn message_has_network_needle(message: &str) -> bool {
pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool { pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
match err { match err {
DiskError::Timeout => true, DiskError::Timeout => true,
DiskError::RemoteClientUnavailable(detail) => message_has_network_needle(detail),
DiskError::Io(io_err) => { DiskError::Io(io_err) => {
if let Some(status) = embedded_tonic_status(io_err) { if let Some(status) = embedded_tonic_status(io_err) {
return is_network_like_status(status); return is_network_like_status(status);
@@ -719,6 +720,16 @@ mod tests {
assert!(!is_network_like_disk_error(&DiskError::FileNotFound)); assert!(!is_network_like_disk_error(&DiskError::FileNotFound));
} }
#[test]
fn network_like_disk_error_keeps_typed_client_failures_classified() {
assert!(is_network_like_disk_error(&DiskError::RemoteClientUnavailable(
"transport error: connection refused".to_string()
)));
assert!(!is_network_like_disk_error(&DiskError::RemoteClientUnavailable(
"invalid client credentials".to_string()
)));
}
#[test] #[test]
fn test_signature_interceptor_keeps_auth_headers() { fn test_signature_interceptor_keeps_auth_headers() {
ensure_test_rpc_secret(); ensure_test_rpc_secret();
crates/ecstore/src/cluster/rpc/peer_rest_client.rs
+19 -17
View File
@@ -626,13 +626,13 @@ impl PeerRestClient {
pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> { pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
if self.offline.load(Ordering::Acquire) { if self.offline.load(Ordering::Acquire) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
return Err(Error::other(format!("peer {} is temporarily offline", self.grid_host))); return Err(Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)));
} }
node_service_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) node_service_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| { .map_err(|err| {
let storage_err = Error::other(format!("can not get client, err: {err}")); let storage_err = Error::RemoteClientUnavailable(format!("can not get client, err: {err}"));
if Self::is_network_like_error(&storage_err) { if Self::is_network_like_error(&storage_err) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
} }
@@ -649,13 +649,13 @@ impl PeerRestClient {
> { > {
if self.offline.load(Ordering::Acquire) { if self.offline.load(Ordering::Acquire) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
return Err(Error::other(format!("peer {} is temporarily offline", self.grid_host))); return Err(Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)));
} }
heal_control_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) heal_control_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| { .map_err(|err| {
let storage_err = Error::other(format!("can not get heal control client, err: {err}")); let storage_err = Error::RemoteClientUnavailable(format!("can not get heal control client, err: {err}"));
if Self::is_network_like_error(&storage_err) { if Self::is_network_like_error(&storage_err) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
} }
@@ -668,13 +668,13 @@ impl PeerRestClient {
) -> Result<TierMutationControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> { ) -> Result<TierMutationControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
if self.offline.load(Ordering::Acquire) { if self.offline.load(Ordering::Acquire) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
return Err(Error::other(format!("peer {} is temporarily offline", self.grid_host))); return Err(Error::RemoteClientUnavailable(format!("peer {} is temporarily offline", self.grid_host)));
} }
tier_mutation_control_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) tier_mutation_control_time_out_client(&self.grid_host, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| { .map_err(|err| {
let storage_err = Error::other(format!("can not get tier mutation control client, err: {err}")); let storage_err = Error::RemoteClientUnavailable(format!("can not get tier mutation control client, err: {err}"));
if Self::is_network_like_error(&storage_err) { if Self::is_network_like_error(&storage_err) {
self.mark_offline_and_spawn_recovery(); self.mark_offline_and_spawn_recovery();
} }
@@ -2213,17 +2213,14 @@ fn tier_config_reload_connection_outcome(err: Error) -> TierConfigReloadOutcome
} }
fn is_tier_config_reload_connection_failure(err: &Error) -> bool { fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
let message = err.to_string(); // A bare "unavailable" is only trusted inside the local dial failure from
// A bare "unavailable" is only trusted inside the local dial-failure // `get_client` (typed as RemoteClientUnavailable), never in application text.
// wrapper from `get_client`, never in application text. if let Error::RemoteClientUnavailable(detail) = err
if message && detail.to_ascii_lowercase().contains("unavailable")
.to_ascii_lowercase()
.split_once("can not get client, err:")
.is_some_and(|(_, local_error)| local_error.contains("unavailable"))
{ {
return true; return true;
} }
message_has_network_needle(&message) message_has_network_needle(&err.to_string())
} }
/// Classifies a reload the peer answered but refused to apply. /// Classifies a reload the peer answered but refused to apply.
@@ -3046,17 +3043,22 @@ mod tests {
TierConfigReloadOutcome::Terminal(_) TierConfigReloadOutcome::Terminal(_)
)); ));
assert!(matches!( assert!(matches!(
tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")), tier_config_reload_connection_outcome(Error::RemoteClientUnavailable("connection unavailable".to_string())),
TierConfigReloadOutcome::TransientReconnect(_) TierConfigReloadOutcome::TransientReconnect(_)
)); ));
// The bare word is trusted only to the right of the dial-failure // The bare word is trusted only inside the typed local dial failure,
// prefix, not anywhere in the message. // not anywhere in application text — including text that mimics the
// old "can not get client" string form, which is retired.
assert!(matches!( assert!(matches!(
tier_config_reload_connection_outcome(Error::other( tier_config_reload_connection_outcome(Error::other(
"bucket unavailable-logs rejected it, then: can not get client, err: some other reason" "bucket unavailable-logs rejected it, then: can not get client, err: some other reason"
)), )),
TierConfigReloadOutcome::Terminal(_) TierConfigReloadOutcome::Terminal(_)
)); ));
assert!(matches!(
tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")),
TierConfigReloadOutcome::Terminal(_)
));
} }
/// A tier mutation issued while another node restarts must still converge on /// A tier mutation issued while another node restarts must still converge on
crates/ecstore/src/cluster/rpc/peer_s3_client.rs
+1 -1
View File
@@ -1090,7 +1090,7 @@ impl RemotePeerS3Client {
pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> { pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| Error::other(format!("can not get client, err: {err}"))) .map_err(|err| Error::RemoteClientUnavailable(err.to_string()))
} }
/// Start health monitoring for the remote peer /// Start health monitoring for the remote peer
crates/ecstore/src/cluster/rpc/remote_disk.rs
+34 -120
View File
@@ -1379,7 +1379,7 @@ impl RemoteDisk {
let addr = addr.to_string(); let addr = addr.to_string();
let mut client = node_service_time_out_client(&addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) let mut client = node_service_time_out_client(&addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| (Error::other(format!("can not get client, err: {err}")), true))?; .map_err(|err| (Error::RemoteClientUnavailable(err.to_string()), true))?;
let request = Request::new(DiskInfoRequest { let request = Request::new(DiskInfoRequest {
disk: endpoint.to_string(), disk: endpoint.to_string(),
opts, opts,
@@ -1719,7 +1719,7 @@ impl RemoteDisk {
/// recovers even without a background monitor. The recovery monitor's own probe path calls the /// recovers even without a background monitor. The recovery monitor's own probe path calls the
/// client directly and is unaffected. /// client directly and is unaffected.
fn offline_bypass_error(&self) -> Option<Error> { fn offline_bypass_error(&self) -> Option<Error> {
internode_offline_bypass_reason(&self.addr).map(Error::other) internode_offline_bypass_reason(&self.addr).map(Error::RemoteClientUnavailable)
} }
async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> { async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
@@ -1728,7 +1728,7 @@ impl RemoteDisk {
} }
node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| Error::other(format!("can not get client, err: {err}"))) .map_err(|err| Error::RemoteClientUnavailable(err.to_string()))
} }
/// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion). /// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion).
@@ -1745,7 +1745,7 @@ impl RemoteDisk {
ChannelClass::Bulk, ChannelClass::Bulk,
) )
.await .await
.map_err(|err| Error::other(format!("can not get client, err: {err}"))) .map_err(|err| Error::RemoteClientUnavailable(err.to_string()))
} }
async fn disk_ref(&self) -> String { async fn disk_ref(&self) -> String {
@@ -2015,10 +2015,7 @@ impl RemoteDisk {
|| async { || async {
let file_info = compat_json(fi)?; let file_info = compat_json(fi)?;
let file_info_bin = encode_file_info_msgpack(fi)?; let file_info_bin = encode_file_info_msgpack(fi)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(RenameDataRequest { let mut request = Request::new(RenameDataRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
src_volume: src_volume.to_string(), src_volume: src_volume.to_string(),
@@ -2088,10 +2085,7 @@ impl RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let options = serde_json::to_string(&opt)?; let options = serde_json::to_string(&opt)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(DeleteRequest { let mut request = Request::new(DeleteRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2214,10 +2208,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(MakeVolumeRequest { let mut request = Request::new(MakeVolumeRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2253,10 +2244,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(MakeVolumesRequest { let mut request = Request::new(MakeVolumesRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volumes: volumes.iter().map(|s| (*s).to_string()).collect(), volumes: volumes.iter().map(|s| (*s).to_string()).collect(),
@@ -2291,10 +2279,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ListVolumesRequest { let request = Request::new(ListVolumesRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
}); });
@@ -2329,10 +2314,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(StatVolumeRequest { let request = Request::new(StatVolumeRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2368,10 +2350,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(DeleteVolumeRequest { let mut request = Request::new(DeleteVolumeRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2423,10 +2402,7 @@ impl DiskAPI for RemoteDisk {
let file_info = serde_json::to_string(&fi)?; let file_info = serde_json::to_string(&fi)?;
let opts = serde_json::to_string(&opts)?; let opts = serde_json::to_string(&opts)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(DeleteVersionRequest { let mut request = Request::new(DeleteVersionRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2598,10 +2574,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(DeletePathsRequest { let mut request = Request::new(DeletePathsRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2626,10 +2599,7 @@ impl DiskAPI for RemoteDisk {
async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> { async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseRequest { let mut request = Request::new(SnapshotLeaseRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2650,10 +2620,7 @@ impl DiskAPI for RemoteDisk {
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> { async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseRenewRequest { let mut request = Request::new(SnapshotLeaseRenewRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2675,10 +2642,7 @@ impl DiskAPI for RemoteDisk {
async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> { async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseReleaseRequest { let mut request = Request::new(SnapshotLeaseReleaseRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -2718,10 +2682,7 @@ impl DiskAPI for RemoteDisk {
"write_metadata", "write_metadata",
move || async move { move || async move {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(WriteMetadataRequest { let mut request = Request::new(WriteMetadataRequest {
disk, disk,
volume: volume.to_string(), volume: volume.to_string(),
@@ -2753,10 +2714,7 @@ impl DiskAPI for RemoteDisk {
"read_metadata", "read_metadata",
|| async { || async {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ReadMetadataRequest { let request = Request::new(ReadMetadataRequest {
volume: volume.to_string(), volume: volume.to_string(),
path: path.to_string(), path: path.to_string(),
@@ -2798,10 +2756,7 @@ impl DiskAPI for RemoteDisk {
"update_metadata", "update_metadata",
move || async move { move || async move {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(UpdateMetadataRequest { let mut request = Request::new(UpdateMetadataRequest {
disk, disk,
volume: volume.to_string(), volume: volume.to_string(),
@@ -2864,10 +2819,7 @@ impl DiskAPI for RemoteDisk {
let opts_str = opts_str.clone(); let opts_str = opts_str.clone();
let opts_bin = opts_bin.clone(); let opts_bin = opts_bin.clone();
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request_payload_bytes = read_version_attribution_enabled.then(|| { let request_payload_bytes = read_version_attribution_enabled.then(|| {
disk.len() disk.len()
.saturating_add(volume.len()) .saturating_add(volume.len())
@@ -2969,10 +2921,7 @@ impl DiskAPI for RemoteDisk {
move || async move { move || async move {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let disk_len = disk.len(); let disk_len = disk.len();
let mut client = self let mut client = self.get_bulk_client().await?;
.get_bulk_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(BatchReadVersionRequest { let request = Request::new(BatchReadVersionRequest {
disk, disk,
batch_read_version_req, batch_read_version_req,
@@ -3081,10 +3030,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ReadXlRequest { let request = Request::new(ReadXlRequest {
disk, disk,
volume: volume.to_string(), volume: volume.to_string(),
@@ -3129,10 +3075,7 @@ impl DiskAPI for RemoteDisk {
|| async { || async {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ListDirRequest { let request = Request::new(ListDirRequest {
disk, disk,
volume: volume.to_string(), volume: volume.to_string(),
@@ -3393,10 +3336,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(RenameFileRequest { let mut request = Request::new(RenameFileRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
src_volume: src_volume.to_string(), src_volume: src_volume.to_string(),
@@ -3438,10 +3378,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(RenamePartRequest { let mut request = Request::new(RenamePartRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
src_volume: src_volume.to_string(), src_volume: src_volume.to_string(),
@@ -3477,10 +3414,7 @@ impl DiskAPI for RemoteDisk {
) -> Result<()> { ) -> Result<()> {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(PreparePartTransactionRequest { let mut request = Request::new(PreparePartTransactionRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
src_volume: src_volume.to_string(), src_volume: src_volume.to_string(),
@@ -3507,10 +3441,7 @@ impl DiskAPI for RemoteDisk {
async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> Result<()> { async fn settle_part_transaction(&self, volume: &str, path: &str, action: PartTransactionAction) -> Result<()> {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SettlePartTransactionRequest { let mut request = Request::new(SettlePartTransactionRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -3553,10 +3484,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let file_info = serde_json::to_string(&fi)?; let file_info = serde_json::to_string(&fi)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(VerifyFileRequest { let request = Request::new(VerifyFileRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -3594,10 +3522,7 @@ impl DiskAPI for RemoteDisk {
); );
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ReadPartsRequest { let request = Request::new(ReadPartsRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
bucket: bucket.to_string(), bucket: bucket.to_string(),
@@ -3635,10 +3560,7 @@ impl DiskAPI for RemoteDisk {
self.execute_with_timeout( self.execute_with_timeout(
|| async { || async {
let file_info = serde_json::to_string(&fi)?; let file_info = serde_json::to_string(&fi)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(CheckPartsRequest { let request = Request::new(CheckPartsRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
volume: volume.to_string(), volume: volume.to_string(),
@@ -3680,10 +3602,7 @@ impl DiskAPI for RemoteDisk {
let read_multiple_req = compat_json(&req)?; let read_multiple_req = compat_json(&req)?;
let read_multiple_req_bin = encode_msgpack(&req)?; let read_multiple_req_bin = encode_msgpack(&req)?;
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self let mut client = self.get_bulk_client().await?;
.get_bulk_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(ReadMultipleRequest { let request = Request::new(ReadMultipleRequest {
disk, disk,
read_multiple_req, read_multiple_req,
@@ -3728,9 +3647,8 @@ impl DiskAPI for RemoteDisk {
|| async { || async {
let data_len = data.len(); let data_len = data.len();
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self.get_bulk_client().await.map_err(|err| { let mut client = self.get_bulk_client().await.inspect_err(|_| {
crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_error(); crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_error();
Error::other(format!("can not get client, err: {err}"))
})?; })?;
let mut request = Request::new(WriteAllRequest { let mut request = Request::new(WriteAllRequest {
disk, disk,
@@ -3785,9 +3703,8 @@ impl DiskAPI for RemoteDisk {
"read_all", "read_all",
|| async { || async {
let disk = self.disk_ref().await; let disk = self.disk_ref().await;
let mut client = self.get_bulk_client().await.map_err(|err| { let mut client = self.get_bulk_client().await.inspect_err(|_| {
crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error(); crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error();
Error::other(format!("can not get client, err: {err}"))
})?; })?;
let request = Request::new(ReadAllRequest { let request = Request::new(ReadAllRequest {
disk, disk,
@@ -3824,10 +3741,7 @@ impl DiskAPI for RemoteDisk {
"disk_info", "disk_info",
|| async { || async {
let opts = serde_json::to_string(&opts)?; let opts = serde_json::to_string(&opts)?;
let mut client = self let mut client = self.get_client().await?;
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let request = Request::new(DiskInfoRequest { let request = Request::new(DiskInfoRequest {
disk: self.endpoint.to_string(), disk: self.endpoint.to_string(),
opts, opts,
crates/ecstore/src/diagnostics/admin_server_info.rs
+1 -1
View File
@@ -148,7 +148,7 @@ async fn is_server_resolvable(endpoint: &Endpoint) -> Result<()> {
let mut client = node_service_time_out_client(&addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) let mut client = node_service_time_out_client(&addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
.await .await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?; .map_err(|err| Error::RemoteClientUnavailable(err.to_string()))?;
let request = Request::new(PingRequest { let request = Request::new(PingRequest {
version: 1, version: 1,
crates/ecstore/src/disk/error.rs
+91 -4
View File
@@ -157,6 +157,15 @@ pub enum DiskError {
#[error("invalid path")] #[error("invalid path")]
InvalidPath, InvalidPath,
/// Internode RPC client acquisition failed (channel build, auth setup, or
/// the peer is marked offline). The detail is diagnostic only: equality and
/// hashing use the wire code alone, so N disks failing for this same cause
/// land in one `reduce_errs` quorum bucket regardless of per-peer detail
/// (backlog#1845). Keep the detail in the rendered message — substring
/// classifiers (network needles, heal recoverability) read it from there.
#[error("remote rpc client unavailable: {0}")]
RemoteClientUnavailable(String),
} }
impl From<crate::erasure::coding::ErasureConstructionError> for DiskError { impl From<crate::erasure::coding::ErasureConstructionError> for DiskError {
@@ -412,10 +421,10 @@ impl From<tonic::Status> for DiskError {
impl From<rustfs_protos::proto_gen::node_service::Error> for DiskError { impl From<rustfs_protos::proto_gen::node_service::Error> for DiskError {
fn from(e: rustfs_protos::proto_gen::node_service::Error) -> Self { fn from(e: rustfs_protos::proto_gen::node_service::Error) -> Self {
if let Some(err) = DiskError::from_u32(e.code) { if let Some(err) = DiskError::from_u32(e.code) {
if matches!(err, DiskError::Io(_)) { match err {
DiskError::other(e.error_info) DiskError::Io(_) => DiskError::other(e.error_info),
} else { DiskError::RemoteClientUnavailable(_) => DiskError::RemoteClientUnavailable(e.error_info),
err err => err,
} }
} else { } else {
DiskError::other(e.error_info) DiskError::other(e.error_info)
@@ -525,6 +534,7 @@ impl Clone for DiskError {
DiskError::SourceStalled => DiskError::SourceStalled, DiskError::SourceStalled => DiskError::SourceStalled,
DiskError::Timeout => DiskError::Timeout, DiskError::Timeout => DiskError::Timeout,
DiskError::InvalidPath => DiskError::InvalidPath, DiskError::InvalidPath => DiskError::InvalidPath,
DiskError::RemoteClientUnavailable(detail) => DiskError::RemoteClientUnavailable(detail.clone()),
} }
} }
} }
@@ -574,6 +584,7 @@ impl DiskError {
DiskError::SourceStalled => 0x28, DiskError::SourceStalled => 0x28,
DiskError::Timeout => 0x29, DiskError::Timeout => 0x29,
DiskError::InvalidPath => 0x2A, DiskError::InvalidPath => 0x2A,
DiskError::RemoteClientUnavailable(_) => 0x2B,
} }
} }
@@ -621,6 +632,7 @@ impl DiskError {
0x28 => Some(DiskError::SourceStalled), 0x28 => Some(DiskError::SourceStalled),
0x29 => Some(DiskError::Timeout), 0x29 => Some(DiskError::Timeout),
0x2A => Some(DiskError::InvalidPath), 0x2A => Some(DiskError::InvalidPath),
0x2B => Some(DiskError::RemoteClientUnavailable(String::new())),
_ => None, _ => None,
} }
} }
@@ -1120,4 +1132,79 @@ mod tests {
assert!(io_error.to_string().contains(&original_message)); assert!(io_error.to_string().contains(&original_message));
} }
} }
#[test]
fn remote_client_unavailable_buckets_ignore_per_peer_detail() {
// The reason this variant exists (backlog#1845): equality and hashing
// use the wire code alone, so N disks failing because their peer's
// client could not be built land in ONE reduce_errs bucket even though
// each carries different diagnostic detail.
let a = DiskError::RemoteClientUnavailable("connection refused to peer 1".to_string());
let b = DiskError::RemoteClientUnavailable("connection refused to peer 2".to_string());
assert_eq!(a, b);
let errors: Vec<Option<DiskError>> = (0..3)
.map(|peer| Some(DiskError::RemoteClientUnavailable(format!("transport error: peer {peer} unreachable"))))
.collect();
let (count, err) = crate::disk::error_reduce::reduce_errs(&errors, &[]);
assert_eq!(count, 3);
assert!(matches!(err, Some(DiskError::RemoteClientUnavailable(_))));
}
#[test]
fn remote_client_unavailable_display_keeps_detail_for_substring_classifiers() {
// Heal recoverability and the peer network classifiers read needles
// ("transport error", "connection refused", "temporarily offline")
// from the rendered message; the typed variant must keep feeding them.
let err = DiskError::RemoteClientUnavailable("transport error: connection refused".to_string());
let rendered = err.to_string();
assert!(rendered.contains("remote rpc client unavailable"));
assert!(rendered.contains("transport error: connection refused"));
}
#[test]
fn remote_client_unavailable_wire_roundtrip_keeps_variant_and_detail() {
let original = DiskError::RemoteClientUnavailable("dial tcp: connection refused".to_string());
let wire: rustfs_protos::proto_gen::node_service::Error = original.clone().into();
assert_eq!(wire.code, 0x2B);
assert_eq!(wire.error_info, "remote rpc client unavailable: dial tcp: connection refused");
let back: DiskError = wire.into();
// The variant (and therefore quorum bucketing) survives the hop; the
// detail gains the display prefix, mirroring the Io re-wrap behavior.
assert_eq!(back, original);
assert!(matches!(
&back,
DiskError::RemoteClientUnavailable(detail) if detail.contains("dial tcp: connection refused")
));
}
#[test]
fn remote_client_unavailable_survives_layer_and_io_bridges() {
let original = DiskError::RemoteClientUnavailable("handshake timed out".to_string());
let storage: crate::error::StorageError = original.clone().into();
assert!(matches!(
&storage,
crate::error::StorageError::RemoteClientUnavailable(detail) if detail == "handshake timed out"
));
let narrowed: DiskError = storage.into();
assert_eq!(narrowed, original);
assert!(matches!(
&narrowed,
DiskError::RemoteClientUnavailable(detail) if detail == "handshake timed out"
));
let io_err: std::io::Error = original.clone().into();
let recovered: DiskError = io_err.into();
assert_eq!(recovered, original);
let cloned = original.clone();
assert!(matches!(
&cloned,
DiskError::RemoteClientUnavailable(detail) if detail == "handshake timed out"
));
assert_eq!(cloned, original);
}
} }
crates/ecstore/src/error/mod.rs
+12
View File
@@ -221,6 +221,13 @@ pub enum StorageError {
Io(#[source] std::io::Error), Io(#[source] std::io::Error),
#[error("Lock error: {0}")] #[error("Lock error: {0}")]
Lock(#[from] rustfs_lock::LockError), Lock(#[from] rustfs_lock::LockError),
/// Internode RPC client acquisition failed. Mirrors
/// `DiskError::RemoteClientUnavailable`: the detail is diagnostic only and
/// excluded from equality/hashing so same-cause failures bucket together
/// during quorum aggregation (backlog#1845).
#[error("remote rpc client unavailable: {0}")]
RemoteClientUnavailable(String),
} }
impl From<crate::erasure::coding::ErasureConstructionError> for StorageError { impl From<crate::erasure::coding::ErasureConstructionError> for StorageError {
@@ -315,6 +322,7 @@ impl From<DiskError> for StorageError {
DiskError::SourceStalled => StorageError::SourceStalled, DiskError::SourceStalled => StorageError::SourceStalled,
DiskError::Timeout => StorageError::Timeout, DiskError::Timeout => StorageError::Timeout,
DiskError::InvalidPath => StorageError::InvalidPath, DiskError::InvalidPath => StorageError::InvalidPath,
DiskError::RemoteClientUnavailable(detail) => StorageError::RemoteClientUnavailable(detail),
} }
} }
} }
@@ -366,6 +374,7 @@ impl From<StorageError> for DiskError {
StorageError::VolumeNotEmpty => DiskError::VolumeNotEmpty, StorageError::VolumeNotEmpty => DiskError::VolumeNotEmpty,
StorageError::VolumeAccessDenied => DiskError::VolumeAccessDenied, StorageError::VolumeAccessDenied => DiskError::VolumeAccessDenied,
StorageError::FileAccessDenied => DiskError::FileAccessDenied, StorageError::FileAccessDenied => DiskError::FileAccessDenied,
StorageError::RemoteClientUnavailable(detail) => DiskError::RemoteClientUnavailable(detail),
_ => DiskError::other(val), _ => DiskError::other(val),
} }
} }
@@ -558,6 +567,7 @@ impl Clone for StorageError {
current: *current, current: *current,
limit: *limit, limit: *limit,
}, },
StorageError::RemoteClientUnavailable(detail) => StorageError::RemoteClientUnavailable(detail.clone()),
} }
} }
} }
@@ -646,6 +656,7 @@ impl StorageError {
StorageError::InvalidPartNumber(_) => StorageErrorCode::InvalidPartNumber, StorageError::InvalidPartNumber(_) => StorageErrorCode::InvalidPartNumber,
StorageError::NamespaceLockQuorumUnavailable { .. } => StorageErrorCode::NamespaceLockQuorumUnavailable, StorageError::NamespaceLockQuorumUnavailable { .. } => StorageErrorCode::NamespaceLockQuorumUnavailable,
StorageError::QuotaExceeded { .. } => StorageErrorCode::QuotaExceeded, StorageError::QuotaExceeded { .. } => StorageErrorCode::QuotaExceeded,
StorageError::RemoteClientUnavailable(_) => StorageErrorCode::RemoteClientUnavailable,
} }
} }
@@ -775,6 +786,7 @@ impl StorageError {
current: Default::default(), current: Default::default(),
limit: Default::default(), limit: Default::default(),
}), }),
StorageErrorCode::RemoteClientUnavailable => Some(StorageError::RemoteClientUnavailable(Default::default())),
} }
} }
} }
crates/storage-api/src/error.rs
+4
View File
@@ -104,6 +104,7 @@ pub enum StorageErrorCode {
Timeout, Timeout,
InvalidPath, InvalidPath,
QuotaExceeded, QuotaExceeded,
RemoteClientUnavailable,
} }
impl StorageErrorCode { impl StorageErrorCode {
@@ -190,6 +191,7 @@ impl StorageErrorCode {
Self::Timeout => 0x51, Self::Timeout => 0x51,
Self::InvalidPath => 0x52, Self::InvalidPath => 0x52,
Self::QuotaExceeded => 0x53, Self::QuotaExceeded => 0x53,
Self::RemoteClientUnavailable => 0x54,
} }
} }
@@ -276,6 +278,7 @@ impl StorageErrorCode {
0x51 => Some(Self::Timeout), 0x51 => Some(Self::Timeout),
0x52 => Some(Self::InvalidPath), 0x52 => Some(Self::InvalidPath),
0x53 => Some(Self::QuotaExceeded), 0x53 => Some(Self::QuotaExceeded),
0x54 => Some(Self::RemoteClientUnavailable),
_ => None, _ => None,
} }
} }
@@ -351,6 +354,7 @@ mod tests {
(StorageErrorCode::OperationCanceled, 0x41), (StorageErrorCode::OperationCanceled, 0x41),
(StorageErrorCode::NamespaceLockQuorumUnavailable, 0x42), (StorageErrorCode::NamespaceLockQuorumUnavailable, 0x42),
(StorageErrorCode::QuotaExceeded, 0x53), (StorageErrorCode::QuotaExceeded, 0x53),
(StorageErrorCode::RemoteClientUnavailable, 0x54),
]; ];
const DISK_PRESERVATION_ERROR_CODES: &[(StorageErrorCode, u32)] = &[ const DISK_PRESERVATION_ERROR_CODES: &[(StorageErrorCode, u32)] = &[
scripts/error-other-format-baseline.txt
+3 -4
View File
@@ -26,16 +26,15 @@
5|crates/ecstore/src/client/transition_api.rs 5|crates/ecstore/src/client/transition_api.rs
3|crates/ecstore/src/cluster/rpc/http_auth.rs 3|crates/ecstore/src/cluster/rpc/http_auth.rs
2|crates/ecstore/src/cluster/rpc/internode_data_transport.rs 2|crates/ecstore/src/cluster/rpc/internode_data_transport.rs
17|crates/ecstore/src/cluster/rpc/peer_rest_client.rs 11|crates/ecstore/src/cluster/rpc/peer_rest_client.rs
11|crates/ecstore/src/cluster/rpc/peer_s3_client.rs 10|crates/ecstore/src/cluster/rpc/peer_s3_client.rs
43|crates/ecstore/src/cluster/rpc/remote_disk.rs 10|crates/ecstore/src/cluster/rpc/remote_disk.rs
7|crates/ecstore/src/config/com.rs 7|crates/ecstore/src/config/com.rs
14|crates/ecstore/src/config/storageclass.rs 14|crates/ecstore/src/config/storageclass.rs
185|crates/ecstore/src/core/pools.rs 185|crates/ecstore/src/core/pools.rs
8|crates/ecstore/src/data_movement/mod.rs 8|crates/ecstore/src/data_movement/mod.rs
2|crates/ecstore/src/data_usage/local_snapshot.rs 2|crates/ecstore/src/data_usage/local_snapshot.rs
12|crates/ecstore/src/data_usage/mod.rs 12|crates/ecstore/src/data_usage/mod.rs
1|crates/ecstore/src/diagnostics/admin_server_info.rs
5|crates/ecstore/src/disk/local.rs 5|crates/ecstore/src/disk/local.rs
1|crates/ecstore/src/disk/mod.rs 1|crates/ecstore/src/disk/mod.rs
5|crates/ecstore/src/erasure/codec/bridge.rs 5|crates/ecstore/src/erasure/codec/bridge.rs