RSDK-858 - Python assert error hot fix (#24)

stuqdog · web-flow · commit 6bbdba44dc19 · 2023-01-17T18:35:58.000-05:00
diff --git a/examples/src/echo/main.rs b/examples/src/echo/main.rs
@@ -9,13 +9,14 @@ use viam::rpc::dial;
 /// Tests unary, server, and bidi streaming with simple echo requests. To run, simply
 /// update the credentials and uri as necessary.
 async fn main() -> Result<()> {
-    let creds = dial::CredentialsExt::new(
+    let creds = dial::RPCCredentials::new(
+        None,
         "robot-location-secret".to_string(),
-        "ytexnwei4fu1xv9csoqxfv4ckl3htsb49mzzey5t15xo9swy".to_string(),
+        "<your secret here>".to_string(),
     );
 
     let c = dial::DialOptions::builder()
-        .uri("webrtc-test-main.jkek76kqnh.viam.cloud")
+        .uri("<your robot address here>")
         .with_credentials(creds)
         .allow_downgrade()
         .connect()
diff --git a/src/rpc/client_channel.rs b/src/rpc/client_channel.rs
@@ -11,7 +11,7 @@ use std::{
     fmt::Debug,
     sync::{
         atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering},
-        Arc,
+        Arc, RwLock,
     },
 };
 use webrtc::{
@@ -28,6 +28,8 @@ pub struct WebRTCClientChannel {
     stream_id_counter: AtomicU64,
     pub(crate) streams: CHashMap<u64, WebRTCClientStream>,
     pub(crate) receiver_bodies: CHashMap<u64, hyper::Body>,
+    // String type rather than error type because anyhow::Error does not derive clone
+    pub(crate) error: RwLock<Option<String>>,
 }
 
 impl Debug for WebRTCClientChannel {
@@ -64,7 +66,9 @@ impl WebRTCClientChannel {
         data_channel: Arc<RTCDataChannel>,
     ) -> Arc<Self> {
         let base_channel = WebRTCBaseChannel::new(peer_connection, data_channel.clone()).await;
+        let error = RwLock::new(None);
         let channel = Self {
+            error,
             base_channel,
             streams: CHashMap::new(),
             stream_id_counter: AtomicU64::new(0),
@@ -84,8 +88,14 @@ impl WebRTCClientChannel {
                         return;
                     }
                 };
-                if let Err(e) = channel.on_channel_message(msg).await {
-                    log::error!("error deserializing message: {e}");
+                let maybe_err = channel.on_channel_message(msg).await;
+                let mut err = channel.error.write().unwrap();
+                match maybe_err {
+                    Err(e) => {
+                        log::error!("error deserializing message: {e}");
+                        *err = Some(e.to_string());
+                    }
+                    Ok(()) => *err = None,
                 }
             })
         }));
@@ -119,7 +129,6 @@ impl WebRTCClientChannel {
 
     async fn on_channel_message(&self, msg: DataChannelMessage) -> Result<()> {
         let response = Response::decode(&*msg.data.to_vec())?;
-        let should_drop_stream = matches!(response.r#type, Some(RespType::Trailers(_)));
         let (active_stream, stream_id) = match response.stream.as_ref() {
             None => {
                 log::error!(
@@ -141,17 +150,17 @@ impl WebRTCClientChannel {
             }
         };
 
-        match active_stream {
-            Ok(mut active_stream) => active_stream.on_response(response).await?,
-            Err(e) => {
-                log::error!("Error acquiring active stream: {e}");
-                return Ok(());
-            }
+        let should_drop_stream = matches!(response.r#type, Some(RespType::Trailers(_)));
+
+        let maybe_err = match active_stream {
+            Ok(mut active_stream) => active_stream.on_response(response).await,
+            Err(e) => Err(anyhow::anyhow!("Error acquiring active stream: {e}")),
         };
+
         if should_drop_stream {
             self.streams.remove(&stream_id);
         }
-        Ok(())
+        maybe_err
     }
 
     pub(crate) fn resp_body_from_stream(&self, stream_id: u64) -> Result<Body> {
diff --git a/src/rpc/client_stream.rs b/src/rpc/client_stream.rs
@@ -44,7 +44,7 @@ impl WebRTCClientStream {
         Ok(())
     }
 
-    async fn process_trailers(&mut self, trailers: ResponseTrailers) {
+    async fn process_trailers(&mut self, trailers: ResponseTrailers) -> Result<()> {
         let trailers_to_send = trailers_from_proto(trailers.clone());
         if let Err(e) = self
             .base_stream
@@ -69,7 +69,14 @@ impl WebRTCClientStream {
             }
         };
 
-        self.base_stream.close_with_recv_error(&mut err.as_ref())
+        self.base_stream.close_with_recv_error(&mut err.as_ref());
+        match err {
+            None => Ok(()),
+            Some(err) => {
+                log::error!("Error processing trailers: {err}");
+                Err(err)
+            }
+        }
     }
 
     // processes response.
@@ -107,10 +114,7 @@ impl WebRTCClientStream {
                 self.process_message(message.to_owned()).await
             }
 
-            Some(Type::Trailers(trailers)) => {
-                self.process_trailers(trailers.to_owned()).await;
-                Ok(())
-            }
+            Some(Type::Trailers(trailers)) => self.process_trailers(trailers.to_owned()).await,
             None => Ok(()),
         }
     }
diff --git a/src/rpc/dial.rs b/src/rpc/dial.rs
@@ -129,6 +129,20 @@ impl Service<http::Request<BoxBody>> for ViamChannel {
                         }
                     };
 
+                    // TODO(RSDK-654) if and when we need to provide better support for
+                    // bidi streaming, we'll need to find a better solution than this for
+                    // handling empty messages
+                    let mut body = hyper::body::to_bytes(body).await.unwrap().to_vec();
+                    if body.is_empty() {
+                        // the body is empty if we make a call that returns an error on the RDK
+                        // side. Python's grpc library expects to see messages with, at a minimum,
+                        // a grpc header, so returning an empty body leads to error. In such a
+                        // case, we instead return a body that just consists of empty header bytes
+                        // to indicate an empty message.
+                        body = vec![0, 0, 0, 0, 0];
+                        status_code = STATUS_CODE_UNKNOWN;
+                    }
+                    let grpc_message = channel.error.read().unwrap().clone();
                     let response = http::response::Response::builder()
                         // standardized gRPC headers.
                         .header("content-type", "application/grpc")
@@ -137,9 +151,13 @@ impl Service<http::Request<BoxBody>> for ViamChannel {
                         .header(TRAILER, "Grpc-Message")
                         .header(TRAILER, "Grpc-Status-Details-Bin")
                         .header("grpc-status", &status_code.to_string())
-                        .version(Version::HTTP_2)
-                        .body(body)
-                        .unwrap();
+                        .version(Version::HTTP_2);
+
+                    let response = match grpc_message {
+                        None => response,
+                        Some(message) => response.header("grpc-message", message),
+                    };
+                    let response = response.body(Body::from(body)).unwrap();
                     Ok(response)
                 };
                 Box::pin(fut)
diff --git a/src/rpc/webrtc.rs b/src/rpc/webrtc.rs
@@ -333,6 +333,15 @@ pub(crate) fn trailers_from_proto(proto: ResponseTrailers) -> HeaderMap {
         None => "0".to_string(),
     };
 
+    match proto.status {
+        Some(ref status) => {
+            let key = HeaderName::from_str("Grpc-Message").unwrap();
+            let val = HeaderValue::from_str(&status.message).unwrap();
+            trailers.insert(key, val);
+        }
+        None => (),
+    }
+
     let k = match HeaderName::from_str(status_name) {
         Ok(k) => k,
         Err(e) => {

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ impl WebRTCClientStream {`
`44`	`44`	`Ok(())`
`45`	`45`	`}`
`46`	`46`
`47`		`- async fn process_trailers(&mut self, trailers: ResponseTrailers) {`
	`47`	`+ async fn process_trailers(&mut self, trailers: ResponseTrailers) -> Result<()> {`
`48`	`48`	`let trailers_to_send = trailers_from_proto(trailers.clone());`
`49`	`49`	`if let Err(e) = self`
`50`	`50`	`.base_stream`
`@@ -69,7 +69,14 @@ impl WebRTCClientStream {`
`69`	`69`	`}`
`70`	`70`	`};`
`71`	`71`
`72`		`- self.base_stream.close_with_recv_error(&mut err.as_ref())`
	`72`	`+ self.base_stream.close_with_recv_error(&mut err.as_ref());`
	`73`	`+ match err {`
	`74`	`+ None => Ok(()),`
	`75`	`+ Some(err) => {`
	`76`	`+ log::error!("Error processing trailers: {err}");`
	`77`	`+ Err(err)`
	`78`	`+ }`
	`79`	`+ }`
`73`	`80`	`}`
`74`	`81`
`75`	`82`	`// processes response.`
`@@ -107,10 +114,7 @@ impl WebRTCClientStream {`
`107`	`114`	`self.process_message(message.to_owned()).await`
`108`	`115`	`}`
`109`	`116`
`110`		`- Some(Type::Trailers(trailers)) => {`
`111`		`- self.process_trailers(trailers.to_owned()).await;`
`112`		`- Ok(())`
`113`		`- }`
	`117`	`+ Some(Type::Trailers(trailers)) => self.process_trailers(trailers.to_owned()).await,`
`114`	`118`	`None => Ok(()),`
`115`	`119`	`}`
`116`	`120`	`}`