From 5c35bef33005505a175b40c4cc50c0cbfd3d2c3b Mon Sep 17 00:00:00 2001 From: WiseDev <83840010+wisedevik@users.noreply.github.com> Date: Fri, 28 Aug 2026 23:43:47 +0300 Subject: [PATCH] send service errors back over rpc instead of dropping the connection the call outcome now rides the frame as Result, so a rejection reaches the caller with its message and the pooled connection stays up. only a real transport failure closes it now; a malformed request comes back as a rejection too. --- crates/service-rpc/src/client.rs | 7 ++-- crates/service-rpc/src/error.rs | 26 ++++++++++++++ crates/service-rpc/src/lib.rs | 2 +- crates/service-rpc/src/server.rs | 10 +++--- crates/service-rpc/tests/error_channel.rs | 43 +++++++++++++++++++++++ 5 files changed, 81 insertions(+), 7 deletions(-) create mode 100644 crates/service-rpc/tests/error_channel.rs diff --git a/crates/service-rpc/src/client.rs b/crates/service-rpc/src/client.rs index 58563b6..b737d0e 100644 --- a/crates/service-rpc/src/client.rs +++ b/crates/service-rpc/src/client.rs @@ -1,4 +1,4 @@ -use crate::error::{RpcError, RpcResult}; +use crate::error::{RpcError, RpcResult, WireError}; use crate::server::{read_frame, write_frame}; use serde::{de::DeserializeOwned, Serialize}; use std::marker::PhantomData; @@ -56,7 +56,10 @@ where read_frame(stream).await }; match exchange.await { - Ok(payload) => return Ok(postcard::from_bytes(&payload)?), + Ok(payload) => { + let outcome: Result = postcard::from_bytes(&payload)?; + return outcome.map_err(RpcError::from); + } Err(error) => { *guard = None; last_error = Some(error); diff --git a/crates/service-rpc/src/error.rs b/crates/service-rpc/src/error.rs index 2de5692..51beed8 100644 --- a/crates/service-rpc/src/error.rs +++ b/crates/service-rpc/src/error.rs @@ -1,3 +1,4 @@ +use serde::{Deserialize, Serialize}; pub type RpcResult = Result; #[derive(Debug, thiserror::Error)] pub enum RpcError { @@ -14,3 +15,28 @@ pub enum RpcError { #[error("service is unavailable: {0}")] Unavailable(String), } + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] +pub enum WireError { + #[error("{0}")] + Rejected(String), + #[error("{0}")] + Unavailable(String), +} +impl From for WireError { + fn from(error: RpcError) -> Self { + match error { + RpcError::Rejected(message) => WireError::Rejected(message), + RpcError::Unavailable(message) => WireError::Unavailable(message), + other => WireError::Unavailable(other.to_string()), + } + } +} +impl From for RpcError { + fn from(error: WireError) -> Self { + match error { + WireError::Rejected(message) => RpcError::Rejected(message), + WireError::Unavailable(message) => RpcError::Unavailable(message), + } + } +} diff --git a/crates/service-rpc/src/lib.rs b/crates/service-rpc/src/lib.rs index 66d7450..c1ee106 100644 --- a/crates/service-rpc/src/lib.rs +++ b/crates/service-rpc/src/lib.rs @@ -6,7 +6,7 @@ pub mod server; pub mod wire; pub use auth::{AuthApi, AuthRequest, AuthResponse, DeviceInfo, LoginOutcome, Session}; pub use client::RpcClient; -pub use error::{RpcError, RpcResult}; +pub use error::{RpcError, RpcResult, WireError}; pub use game::{AccountRef, GameApi, GameRequest, GameResponse, HomeRequestKind}; pub use server::{serve, RpcService}; pub use wire::WireMessage; diff --git a/crates/service-rpc/src/server.rs b/crates/service-rpc/src/server.rs index cf25a54..9ae30be 100644 --- a/crates/service-rpc/src/server.rs +++ b/crates/service-rpc/src/server.rs @@ -1,4 +1,4 @@ -use crate::error::{RpcError, RpcResult}; +use crate::error::{RpcError, RpcResult, WireError}; use serde::{de::DeserializeOwned, Serialize}; use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -31,9 +31,11 @@ async fn handle_connection(mut stream: TcpStream, service: Arc Err(RpcError::Closed) => return Ok(()), Err(error) => return Err(error), }; - let decoded: S::Request = postcard::from_bytes(&request)?; - let response = service.call(decoded).await?; - let encoded = postcard::to_allocvec(&response)?; + let outcome: Result = match postcard::from_bytes::(&request) { + Ok(decoded) => service.call(decoded).await.map_err(WireError::from), + Err(error) => Err(WireError::Rejected(format!("undecodable request: {error}"))), + }; + let encoded = postcard::to_allocvec(&outcome)?; write_frame(&mut stream, &encoded).await?; } } diff --git a/crates/service-rpc/tests/error_channel.rs b/crates/service-rpc/tests/error_channel.rs new file mode 100644 index 0000000..f2882c1 --- /dev/null +++ b/crates/service-rpc/tests/error_channel.rs @@ -0,0 +1,43 @@ +use std::sync::Arc; +use service_rpc::{serve, RpcClient, RpcError, RpcResult, RpcService}; +use tokio::net::TcpListener; + +struct Doubler; + +#[async_trait::async_trait] +impl RpcService for Doubler { + type Request = i32; + type Response = i32; + fn service_name(&self) -> &'static str { + "doubler" + } + async fn call(&self, request: i32) -> RpcResult { + if request == 0 { + return Err(RpcError::Rejected("zero is not allowed".to_owned())); + } + Ok(request * 2) + } +} + +#[tokio::test] +async fn a_service_error_reaches_the_caller_and_keeps_the_connection() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let endpoint = listener.local_addr().expect("addr").to_string(); + tokio::spawn(async move { + let _ = serve(listener, Arc::new(Doubler)).await; + }); + + let client: RpcClient = RpcClient::with_pool_size(endpoint, 1); + + assert_eq!(client.call(&5).await.expect("ok call"), 10); + + match client.call(&0).await { + Err(RpcError::Rejected(message)) => assert_eq!(message, "zero is not allowed"), + other => panic!("expected a rejection, got {other:?}"), + } + + assert_eq!( + client.call(&7).await.expect("connection survived the error"), + 14 + ); +}