feat: support downcast WorkerQuitReason::Fatal

Author: 4t145

crates/rmcp/src/transport/common/reqwest/streamable_http_client.rs

@@ -36,7 +36,7 @@ impl StreamableHttpClient for reqwest::Client {
         }
         let response = request_builder.send().await?;
         if response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED {
-            return Err(StreamableHttpError::SeverDoesNotSupportSse);
+            return Err(StreamableHttpError::ServerDoesNotSupportSse);
         }
         let response = response.error_for_status()?;
         match response.headers().get(reqwest::header::CONTENT_TYPE) {

crates/rmcp/src/transport/streamable_http_client.rs

@@ -33,15 +33,17 @@ pub enum StreamableHttpError<E: std::error::Error + Send + Sync + 'static> {
     #[error("Unexpected content type: {0:?}")]
     UnexpectedContentType(Option<String>),
     #[error("Server does not support SSE")]
-    SeverDoesNotSupportSse,
+    ServerDoesNotSupportSse,
     #[error("Server does not support delete session")]
-    SeverDoesNotSupportDeleteSession,
+    ServerDoesNotSupportDeleteSession,
     #[error("Tokio join error: {0}")]
     TokioJoinError(#[from] tokio::task::JoinError),
     #[error("Deserialize error: {0}")]
     Deserialize(#[from] serde_json::Error),
     #[error("Transport channel closed")]
     TransportChannelClosed,
+    #[error("Missing session id in response")]
+    MissingSessionIdInResponse,
     #[cfg(feature = "auth")]
     #[cfg_attr(docsrs, doc(cfg(feature = "auth")))]
     #[error("Auth error: {0}")]
@@ -54,6 +56,12 @@ impl From<reqwest::Error> for StreamableHttpError<reqwest::Error> {
     }
 }
 
+#[derive(Debug, Clone, Error)]
+pub enum StreamableHttpProtocolError {
+    #[error("Missing session id in response")]
+    MissingSessionIdInResponse,
+}
+
 pub enum StreamableHttpPostResponse {
     Accepted,
     Json(ServerJsonRpcMessage, Option<String>),
@@ -288,7 +296,7 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
         } else {
             if !self.config.allow_stateless {
                 return Err(WorkerQuitReason::fatal(
-                    "missing session id in initialize response",
+                    StreamableHttpError::<C::Error>::MissingSessionIdInResponse,
                     "process initialize response",
                 ));
             }
@@ -308,7 +316,7 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                     Ok(_) => {
                         tracing::info!(session_id = session_id.as_ref(), "delete session success")
                     }
-                    Err(StreamableHttpError::SeverDoesNotSupportDeleteSession) => {
+                    Err(StreamableHttpError::ServerDoesNotSupportDeleteSession) => {
                         tracing::info!(
                             session_id = session_id.as_ref(),
                             "server doesn't support delete session"
@@ -373,14 +381,14 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                     ));
                     tracing::debug!("got common stream");
                 }
-                Err(StreamableHttpError::SeverDoesNotSupportSse) => {
+                Err(StreamableHttpError::ServerDoesNotSupportSse) => {
                     tracing::debug!("server doesn't support sse, skip common stream");
                 }
                 Err(e) => {
                     // fail to get common stream
                     tracing::error!("fail to get common stream: {e}");
                     return Err(WorkerQuitReason::fatal(
-                        "fail to get general purpose event stream",
+                        e,
                         "get general purpose event stream",
                     ));
                 }

crates/rmcp/src/transport/streamable_http_server/session/local.rs

@@ -317,7 +317,7 @@ enum OutboundChannel {
     RequestWise { id: HttpRequestId, close: bool },
     Common,
 }
-
+#[derive(Debug)]
 pub struct StreamableHttpMessageReceiver {
     pub http_request_id: Option<HttpRequestId>,
     pub inner: Receiver<ServerSseMessage>,
@@ -534,8 +534,8 @@ impl LocalSessionWorker {
         }
     }
 }
-
-enum SessionEvent {
+#[derive(Debug)]
+pub enum SessionEvent {
     ClientMessage {
         message: ClientJsonRpcMessage,
         http_request_id: Option<HttpRequestId>,
@@ -695,6 +695,17 @@ impl LocalSessionHandle {
 
 pub type SessionTransport = WorkerTransport<LocalSessionWorker>;
 
+#[derive(Debug, Error)]
+pub enum LocalSessionError {
+    #[error("transport terminated")]
+    TransportTerminated,
+    #[error("unexpected message: {0:?}")]
+    UnexpectedEvent(SessionEvent),
+    #[error("fail to send initialize request {0}")]
+    FailToSendInitializeRequest(SessionError),
+    #[error("keep alive timeout")]
+    KeepAliveTimeout,
+}
 impl Worker for LocalSessionWorker {
     type Error = SessionError;
     type Role = RoleServer;
@@ -718,11 +729,14 @@ impl Worker for LocalSessionWorker {
         }
         // waiting for initialize request
         let evt = self.event_rx.recv().await.ok_or_else(|| {
-            WorkerQuitReason::fatal("transport terminated", "get initialize request")
+            WorkerQuitReason::fatal(
+                LocalSessionError::TransportTerminated,
+                "get initialize request",
+            )
         })?;
         let SessionEvent::InitializeRequest { request, responder } = evt else {
             return Err(WorkerQuitReason::fatal(
-                "unexpected message",
+                LocalSessionError::UnexpectedEvent(evt),
                 "get initialize request",
             ));
         };
@@ -732,7 +746,9 @@ impl Worker for LocalSessionWorker {
             .send(Ok(send_initialize_response.message))
             .map_err(|_| {
                 WorkerQuitReason::fatal(
-                    "failed to send initialize response to http service",
+                    LocalSessionError::FailToSendInitializeRequest(
+                        SessionError::SessionServiceTerminated,
+                    ),
                     "send initialize response",
                 )
             })?;
@@ -749,7 +765,7 @@ impl Worker for LocalSessionWorker {
                     if let Some(event) = event {
                         InnerEvent::FromHttpService(event)
                     } else {
-                        return Err(WorkerQuitReason::fatal("session dropped", "waiting next session event"))
+                        return Err(WorkerQuitReason::fatal(LocalSessionError::TransportTerminated, "waiting next session event"))
                     }
                 },
                 from_handler = context.recv_from_handler() => {
@@ -759,7 +775,7 @@ impl Worker for LocalSessionWorker {
                     return Err(WorkerQuitReason::Cancelled)
                 }
                 _ = keep_alive_timeout => {
-                    return Err(WorkerQuitReason::fatal("keep live timeout", "poll next session event"))
+                    return Err(WorkerQuitReason::fatal(LocalSessionError::KeepAliveTimeout, "poll next session event"))
                 }
             };
             match event {

crates/rmcp/src/transport/worker.rs

@@ -12,7 +12,7 @@ pub enum WorkerQuitReason {
     Join(#[from] tokio::task::JoinError),
     #[error("Transport fatal {error}, when {context}")]
     Fatal {
-        error: Cow<'static, str>,
+        error: Box<dyn std::error::Error + Send>,
         context: Cow<'static, str>,
     },
     #[error("Transport canncelled")]
@@ -24,17 +24,20 @@ pub enum WorkerQuitReason {
 }
 
 impl WorkerQuitReason {
-    pub fn fatal(msg: impl Into<Cow<'static, str>>, context: impl Into<Cow<'static, str>>) -> Self {
+    pub fn fatal(
+        error: impl std::error::Error + Send + 'static,
+        context: impl Into<Cow<'static, str>>,
+ ) -> Self {
         Self::Fatal {
-            error: msg.into(),
+            error: Box::new(error),
             context: context.into(),
         }
     }
-    pub fn fatal_context<E: std::error::Error>(
+    pub fn fatal_context<E: std::error::Error + Send + 'static>(
         context: impl Into<Cow<'static, str>>,
     ) -> impl FnOnce(E) -> Self {
         |e| Self::Fatal {
-            error: Cow::Owned(format!("{e}")),
+            error: Box::new(e),
             context: context.into(),
         }
     }