handle replica proxy stream error

Author: MarinPostma

sqld/src/rpc/replica_proxy.rs

@@ -39,8 +39,20 @@ impl Proxy for ReplicaProxyService {
         &self,
         req: tonic::Request<tonic::Streaming<ExecReq>>,
     ) -> Result<tonic::Response<Self::StreamExecStream>, tonic::Status> {
-        let (meta, ext, stream) = req.into_parts();
-        let mut req = tonic::Request::from_parts(meta, ext, stream.map(|r| r.unwrap())); // TODO: handle mapping error
+        let (meta, ext, mut stream) = req.into_parts();
+        let stream = async_stream::stream! {
+            while let Some(it) = stream.next().await {
+                match it {
+                    Ok(it) => yield it,
+                    Err(e) => {
+                        // close the stream on error
+                        tracing::error!("error proxying stream request: {e}");
+                        break
+                    },
+                }
+            }
+        };
+        let mut req = tonic::Request::from_parts(meta, ext, stream);
         self.do_auth(&mut req)?;
         let mut client = self.client.clone();
         client.stream_exec(req).await

sqld/tests/cluster/mod.rs

@@ -2,9 +2,7 @@
 
 use super::common;
 
-use insta::assert_snapshot;
 use libsql::{Database, Value};
-use serde_json::json;
 use sqld::config::{AdminApiConfig, RpcClientConfig, RpcServerConfig, UserApiConfig};
 use tempfile::tempdir;
 use tokio::{task::JoinSet, time::Duration};

sqld/tests/namespaces/mod.rs

@@ -4,6 +4,7 @@ use std::path::PathBuf;
 
 use crate::common::http::Client;
 use crate::common::net::{init_tracing, TestServer, TurmoilAcceptor, TurmoilConnector};
+use insta::assert_snapshot;
 use libsql::{Database, Value};
 use serde_json::json;
 use sqld::config::{AdminApiConfig, RpcServerConfig, UserApiConfig};
@@ -44,7 +45,8 @@ fn make_primary(sim: &mut Sim, path: PathBuf) {
 #[test]
 fn create_namespace() {
     let mut sim = Builder::new().build();
-    make_cluster(&mut sim, 0, false);
+    let tmp = tempdir().unwrap();
+    make_primary(&mut sim, tmp.path().into());
 
     sim.client("client", async {
         let db =