send service errors back over rpc instead of dropping the connection

the call outcome now rides the frame as Result<Response, WireError>, 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

@@ -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<Res, WireError> = postcard::from_bytes(&payload)?;
+                    return outcome.map_err(RpcError::from);
+                }
                 Err(error) => {
                     *guard = None;
                     last_error = Some(error);

crates/service-rpc/src/error.rs

@@ -1,3 +1,4 @@
+use serde::{Deserialize, Serialize};
 pub type RpcResult<T> = Result<T, RpcError>;
 #[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<RpcError> 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<WireError> for RpcError {
+    fn from(error: WireError) -> Self {
+        match error {
+            WireError::Rejected(message) => RpcError::Rejected(message),
+            WireError::Unavailable(message) => RpcError::Unavailable(message),
+        }
+    }
+}

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;

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<S: RpcService>(mut stream: TcpStream, service: Arc<S>
             Err(RpcError::Closed) => return Ok(()),
             Err(error) => return Err(error),
         };
-        let decoded: S::Request = postcard::from_bytes(&request)?;
+        let outcome: Result<S::Response, WireError> = match postcard::from_bytes::<S::Request>(&request) {
-        let response = service.call(decoded).await?;
+            Ok(decoded) => service.call(decoded).await.map_err(WireError::from),
-        let encoded = postcard::to_allocvec(&response)?;
+            Err(error) => Err(WireError::Rejected(format!("undecodable request: {error}"))),
+        };
+        let encoded = postcard::to_allocvec(&outcome)?;
         write_frame(&mut stream, &encoded).await?;
     }
 }

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<i32> {
+        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<i32, i32> = 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
+    );
+}