Repository: sugyan/atrium (language: rust)

Issue:
`atrium_xrpc_server` isn't compatible with `tokio::spawn`
`atrium_xrpc_server::stream::frames::Frame::try_from` returns `Result<Self, Box<dyn std::error::Error>>`. The problem is that `dyn std::error::Error` is not `Send`, which means that if the result of `Frame::try_from` is potentially used across an await, the entire future is not `Send`.

To reproduce this, modify the firehose example to spawn the firehose running method instead of running it directly with `#[tokio::main]`. I tested this at commit 7d99ccd36c8bb562276732d1b9bb888f42530151.

It would be helpful if the error type was `Send + Sync + 'static`, to also make it compatible with the `anyhow` crate. But that can be handled by rewriting the error as a string, whereas the future not being `Send` is more problematic for usage.


Discussion / hints:

Fix patch:
diff --git a/atrium-xrpc-server/Cargo.toml b/atrium-xrpc-server/Cargo.toml
index 1f47d115..0269c490 100644
--- a/atrium-xrpc-server/Cargo.toml
+++ b/atrium-xrpc-server/Cargo.toml
@@ -7,5 +7,6 @@ edition = "2021"
 
 [dependencies]
 atrium-api = "0.3"
+anyhow = "1.0.71"
 libipld-core = { version = "0.16.0", features = ["serde-codec"] }
 serde_ipld_dagcbor = "0.3.0"
diff --git a/atrium-xrpc-server/src/stream/frames.rs b/atrium-xrpc-server/src/stream/frames.rs
index 1c5dc830..500a72a2 100644
--- a/atrium-xrpc-server/src/stream/frames.rs
+++ b/atrium-xrpc-server/src/stream/frames.rs
@@ -1,20 +1,7 @@
 use atrium_api::com::atproto::sync::subscribe_repos::{Commit, Message};
 use libipld_core::ipld::Ipld;
-use std::error::Error;
-use std::fmt::{Display, Formatter};
 use std::io::Cursor;
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-struct FrameTypeError;
-
-impl Display for FrameTypeError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "invalid frame type")
-    }
-}
-
-impl Error for FrameTypeError {}
-
 // original definition:
 //```
 // export enum FrameType {
@@ -38,7 +25,7 @@ enum FrameHeader {
 }
 
 impl TryFrom<Ipld> for FrameHeader {
-    type Error = FrameTypeError;
+    type Error = anyhow::Error;
 
     fn try_from(value: Ipld) -> Result<Self, <FrameHeader as TryFrom<Ipld>>::Error> {
         if let Ipld::Map(map) = value {
@@ -57,7 +44,7 @@ impl TryFrom<Ipld> for FrameHeader {
                 }
             }
         }
-        Err(FrameTypeError)
+        Err(anyhow::anyhow!("invalid frame type"))
     }
 }
 
@@ -79,7 +66,7 @@ pub struct ErrorFrame {
 }
 
 impl TryFrom<&[u8]> for Frame {
-    type Error = Box<dyn Error>;
+    type Error = anyhow::Error;
 
     fn try_from(value: &[u8]) -> Result<Self, <Frame as TryFrom<&[u8]>>::Error> {
         let mut cursor = Cursor::new(value);
@@ -89,7 +76,7 @@ impl TryFrom<&[u8]> for Frame {
             }
             _ => {
                 // TODO
-                return Err(Box::new(FrameTypeError));
+                return Err(anyhow::anyhow!("invalid frame type"));
             }
         };
         let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(left)?;
@@ -130,9 +117,10 @@ mod tests {
         // {"op": 1, "t": "#commit"}
         let data = serialized_data("a2626f700161746723636f6d6d6974");
         let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
+        let result = FrameHeader::try_from(ipld);
         assert_eq!(
-            FrameHeader::try_from(ipld),
-            Ok(FrameHeader::Message(Some(String::from("#commit"))))
+            result.expect("failed to deserialize"),
+            FrameHeader::Message(Some(String::from("#commit")))
         );
     }
 
@@ -141,7 +129,8 @@ mod tests {
         // {"op": -1}
         let data = serialized_data("a1626f7020");
         let ipld = serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
-        assert_eq!(FrameHeader::try_from(ipld), Ok(FrameHeader::Error));
+        let result = FrameHeader::try_from(ipld);
+        assert_eq!(result.expect("failed to deserialize"), FrameHeader::Error);
     }
 
     #[test]
@@ -151,14 +140,22 @@ mod tests {
             let data = serialized_data("a2626f700261746723636f6d6d6974");
             let ipld =
                 serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
-            assert_eq!(FrameHeader::try_from(ipld), Err(FrameTypeError));
+            let result = FrameHeader::try_from(ipld);
+            assert_eq!(
+                result.expect_err("must be failed").to_string(),
+                "invalid frame type"
+            );
         }
         {
             // {"op": -2}
             let data = serialized_data("a1626f7021");
             let ipld =
                 serde_ipld_dagcbor::from_slice::<Ipld>(&data).expect("failed to deserialize");
-            assert_eq!(FrameHeader::try_from(ipld), Err(FrameTypeError));
+            let result = FrameHeader::try_from(ipld);
+            assert_eq!(
+                result.expect_err("must be failed").to_string(),
+                "invalid frame type"
+            );
         }
     }
 }
diff --git a/examples/firehose/src/main.rs b/examples/firehose/src/main.rs
index 64aedb60..cd694cda 100644
--- a/examples/firehose/src/main.rs
+++ b/examples/firehose/src/main.rs
@@ -15,7 +15,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     Ok(())
 }
 
-async fn process_message(message: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
+async fn process_message(message: &[u8]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     match Frame::try_from(message)? {
         Frame::Message(message) => {
             match message.body {


Test patch: