From f7c4ac11b2d1a80fc850e13f483c35bdc128cc36 Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 6 Oct 2026 08:11:19 -0700 Subject: [PATCH] fix(heal): retry typed RPC transport failures after peer restart (#8365) * fix(heal): retry typed RPC transport failures after peer restart * test(heal): keep original errors in GOAWAY clone coverage --- crates/heal/Cargo.toml | 2 +- crates/heal/src/error.rs | 128 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+), 1 deletion(-) diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index 5cf08360d..a96a88510 100644 --- a/crates/heal/Cargo.toml +++ b/crates/heal/Cargo.toml @@ -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 } diff --git a/crates/heal/src/error.rs b/crates/heal/src/error.rs index a771a6663..b00f0000b 100644 --- a/crates/heal/src/error.rs +++ b/crates/heal/src/error.rs @@ -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::() + .or_else(|| source.source().and_then(|source| source.downcast_ref::())) + }); + 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::())); + } let Some(error) = error.get_ref().and_then(|source| source.downcast_ref::()) 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::()), + "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::()) + .expect("RPC conversion and cloning must retain the typed status"); + assert!( + std::error::Error::source(retained).is_some_and(|source| source.is::()), + "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};