gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 08:02:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): retry typed RPC transport failures after peer restart (#8365)

Browse Source 
* fix(heal): retry typed RPC transport failures after peer restart

* test(heal): keep original errors in GOAWAY clone coverage
This commit is contained in:
Chris
2026-10-06 08:11:19 -07:00
committed by GitHub
parent 9d9e92bab8
commit f7c4ac11b2
2 changed files with 129 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/Cargo.toml
+1 -1
View File
@@ -86,6 +86,7 @@ rustfs-madmin = { workspace = true }
rustfs-utils = { workspace = true }
tokio = { workspace = true, features = ["sync", "io-util", "time", "macros", "fs", "rt-multi-thread"] }
tokio-util = { workspace = true, features = ["io", "compat"] }
tonic = { workspace = true }
tracing = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] }
@@ -100,7 +101,6 @@ crc-fast = { workspace = true }
sha2 = { workspace = true }
[dev-dependencies]
tonic = { workspace = true }
rustfs-filemeta = { workspace = true }
serde_json = { workspace = true, features = ["raw_value"] }
rustfs-test-utils = { workspace = true }
crates/heal/src/error.rs
+128
View File
@@ -221,6 +221,18 @@ fn is_recoverable_internode_error(error: &std::io::Error) -> bool {
if DiskError::io_error_is_rpc_cancelled(error) {
return true;
}
let status = error.get_ref().and_then(|source| {
source
.downcast_ref::<tonic::Status>()
.or_else(|| source.source().and_then(|source| source.downcast_ref::<tonic::Status>()))
});
if let Some(status) = status {
// GOAWAY can surface as Internal/Unknown; replay only when the status
// retains the typed transport cause, within the existing heal budget.
return status.code() == tonic::Code::Unavailable
|| (matches!(status.code(), tonic::Code::Internal | tonic::Code::Unknown)
&& std::error::Error::source(status).is_some_and(|source| source.is::<tonic::transport::Error>()));
}
let Some(error) = error.get_ref().and_then(|source| source.downcast_ref::<InternodeHttpError>()) else {
return false;
};
@@ -346,6 +358,122 @@ mod tests {
}
}
#[tokio::test]
async fn rpc_goaway_is_recoverable_across_storage_error_conversions() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tonic::codegen::Service;
tokio::time::timeout(std::time::Duration::from_secs(5), async {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind HTTP/2 peer");
let address = listener.local_addr().expect("HTTP/2 peer address");
let (release, held) = tokio::sync::oneshot::channel::<()>();
let peer = tokio::spawn(async move {
let (mut stream, _) = listener.accept().await.expect("accept internode channel");
let mut preface = [0_u8; 24];
stream.read_exact(&mut preface).await.expect("read HTTP/2 preface");
assert_eq!(&preface, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n");
stream
.write_all(&[0, 0, 0, 4, 0, 0, 0, 0, 0])
.await
.expect("send HTTP/2 settings");
loop {
let mut header = [0_u8; 9];
stream.read_exact(&mut header).await.expect("read HTTP/2 frame header");
let length = (usize::from(header[0]) << 16) | (usize::from(header[1]) << 8) | usize::from(header[2]);
assert!(length <= 4096, "fixture frame exceeds its budget");
let mut payload = vec![0_u8; length];
stream.read_exact(&mut payload).await.expect("read HTTP/2 frame payload");
if header[3] == 1 {
// A graceful peer restart rejects the admitted stream
// with GOAWAY(NO_ERROR), without an application status.
stream
.write_all(&[0, 0, 8, 7, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])
.await
.expect("send HTTP/2 GOAWAY");
held.await.expect("hold peer until the client observes GOAWAY");
return;
}
}
});
let mut channel = tonic::transport::Endpoint::from_shared(format!("http://{address}"))
.expect("fixture endpoint")
.connect()
.await
.expect("connect internode channel");
std::future::poll_fn(|cx| channel.poll_ready(cx))
.await
.expect("internode channel ready");
let request = http::Request::builder()
.uri("/rustfs.node/HealObject")
.header(http::header::CONTENT_TYPE, "application/grpc")
.body(tonic::body::Body::empty())
.expect("build internode request");
let transport = channel.call(request).await.expect_err("GOAWAY must terminate the RPC");
let status = tonic::Status::from_error(Box::new(transport));
release.send(()).expect("release HTTP/2 peer");
peer.await.expect("HTTP/2 peer completed");
assert_eq!(status.code(), tonic::Code::Internal, "{status:?}");
assert!(
std::error::Error::source(&status).is_some_and(|source| source.is::<tonic::transport::Error>()),
"the real channel error must retain its typed transport source: {status:?}"
);
let disk = DiskError::from(status.clone());
let storage = EcstoreError::from(status.clone());
let disk_storage = EcstoreError::from(DiskError::from(status.clone()));
for error in [
Error::Disk(disk.clone()),
Error::Disk(disk),
Error::Storage(storage.clone()),
Error::Storage(storage),
Error::Storage(disk_storage.clone()),
Error::Storage(disk_storage),
Error::Io(std::io::Error::from(DiskError::from(status))),
] {
let io = match &error {
Error::Disk(DiskError::Io(io)) | Error::Storage(EcstoreError::Io(io)) | Error::Io(io) => io,
_ => panic!("RPC conversion must retain the I/O error: {error:?}"),
};
let retained = io
.get_ref()
.and_then(|source| source.source())
.and_then(|source| source.downcast_ref::<tonic::Status>())
.expect("RPC conversion and cloning must retain the typed status");
assert!(
std::error::Error::source(retained).is_some_and(|source| source.is::<tonic::transport::Error>()),
"RPC conversion and cloning must retain the typed transport source"
);
assert!(
!super::is_recoverable_heal_error_message(&error.to_string()),
"GOAWAY regression must not be covered by the legacy message fallback: {error}"
);
assert!(error.is_recoverable_heal(), "a peer GOAWAY must remain recoverable: {error}");
}
})
.await
.expect("HTTP/2 GOAWAY fixture completed within its budget");
}
#[test]
fn rpc_application_errors_do_not_become_recoverable_from_transport_text() {
for code in [
tonic::Code::Internal,
tonic::Code::Unknown,
tonic::Code::PermissionDenied,
tonic::Code::Unauthenticated,
tonic::Code::DataLoss,
] {
let status = tonic::Status::new(code, "h2 protocol error: http2 error");
for error in [
Error::Disk(DiskError::from(status.clone())),
Error::Storage(EcstoreError::from(status.clone())),
Error::Io(std::io::Error::from(DiskError::from(status.clone()))),
] {
assert!(!error.is_recoverable_heal(), "application RPC errors must stay terminal: {error:?}");
}
}
assert!(!Error::Io(std::io::Error::other("h2 protocol error: http2 error")).is_recoverable_heal());
}
#[tokio::test]
async fn read_http_500_is_not_a_retryable_heal_write() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};