fmt

Author: MarinPostma

sqld/src/connection/write_proxy.rs

@@ -153,7 +153,6 @@ impl WriteProxyConnection {
         F: FnOnce(&mut RemoteConnection) -> BoxFuture<'_, crate::Result<Ret>>,
     {
         let mut remote_conn = self.remote_conn.lock().await;
-        // TODO: catch broken connection, and reset it to None.
         if remote_conn.is_some() {
             cb(remote_conn.as_mut().unwrap()).await
         } else {
@@ -471,7 +470,10 @@ pub mod test {
     use super::*;
     use crate::{
         query_result_builder::{test::test_driver, Column, QueryResultBuilderError},
-        rpc::{proxy::rpc::{query_result::RowResult, ExecuteResults}, streaming_exec::test::random_valid_program_resp},
+        rpc::{
+            proxy::rpc::{query_result::RowResult, ExecuteResults},
+            streaming_exec::test::random_valid_program_resp,
+        },
     };
 
     /// generate an arbitraty rpc value. see build.rs for usage.
@@ -562,6 +564,11 @@ pub mod test {
             builder_config: QueryBuilderConfig::default(),
         };
 
-        remote.execute(Program::seq(&[]), validator).await.unwrap().0.into_ret();
+        remote
+            .execute(Program::seq(&[]), validator)
+            .await
+            .unwrap()
+            .0
+            .into_ret();
     }
 }

sqld/src/http/user/result_builder.rs

@@ -312,7 +312,7 @@ impl QueryResultBuilder for JsonHttpPayloadBuilder {
 #[cfg(test)]
 mod test {
 
-    use crate::query_result_builder::test::{random_transition, fsm_builder_driver};
+    use crate::query_result_builder::test::{fsm_builder_driver, random_transition};
 
     use super::*;
 

sqld/src/lib.rs

@@ -500,6 +500,8 @@ where
             let proxy_service =
                 ProxyService::new(namespaces.clone(), None, self.disable_namespaces);
             // Garbage collect proxy clients every 30 seconds
+            // TODO: this will no longer be necessary once client have adopted the streaming proxy
+            // protocol
             self.join_set.spawn({
                 let clients = proxy_service.clients();
                 async move {

sqld/src/query_result_builder.rs

@@ -715,7 +715,7 @@ pub mod test {
 
             trace.push(state);
             if state == FsmState::Finish {
-                break
+                break;
             }
 
             max_steps = max_steps.saturating_sub(1);
@@ -818,7 +818,10 @@ pub mod test {
             Ok(())
         }
 
-        fn step_error(&mut self, _error: crate::error::Error) -> Result<(), QueryResultBuilderError> {
+        fn step_error(
+            &mut self,
+            _error: crate::error::Error,
+        ) -> Result<(), QueryResultBuilderError> {
             assert_eq!(self.trace[self.current], FsmState::StepError);
             self.current += 1;
             Ok(())

sqld/src/rpc/streaming_exec.rs

@@ -365,7 +365,9 @@ pub mod test {
     use crate::auth::{Authorized, Permission};
     use crate::connection::libsql::LibSqlConnection;
     use crate::connection::program::Program;
-    use crate::query_result_builder::test::{TestBuilder, ValidateTraceBuilder, random_transition, fsm_builder_driver};
+    use crate::query_result_builder::test::{
+        fsm_builder_driver, random_transition, TestBuilder, ValidateTraceBuilder,
+    };
     use crate::rpc::proxy::rpc::StreamProgramReq;
 
     use super::*;
@@ -491,61 +493,6 @@ pub mod test {
         assert_debug_snapshot!(builder.into_ret());
     }
 
-    #[tokio::test]
-    async fn interupt_query() {
-        let tmp = tempdir().unwrap();
-        let conn = LibSqlConnection::new_test(tmp.path());
-        let (snd, rcv) = mpsc::channel(1);
-        let auth = Authenticated::Authorized(Authorized {
-            namespace: None,
-            permission: Permission::FullAccess,
-        });
-        let stream = make_proxy_stream(conn, auth, ReceiverStream::new(rcv));
-
-        pin!(stream);
-
-        let req = exec_req_stmt("create table test (foo)", 0);
-        snd.send(Ok(req)).await.unwrap();
-        let resp = stream.next().await.unwrap().unwrap();
-        assert_eq!(resp.request_id, 0);
-        for i in 1..50 {
-            let req = exec_req_stmt(
-                r#"insert into test values ("something moderately long")"#,
-                i,
-            );
-            snd.send(Ok(req)).await.unwrap();
-            let resp = stream.next().await.unwrap().unwrap();
-            assert_eq!(resp.request_id, i);
-        }
-
-        let req = exec_req_stmt("select * from test", 100);
-        snd.send(Ok(req)).await.unwrap();
-
-        let mut num_resp = 0;
-        let mut builder = TestBuilder::default();
-        loop {
-            let Response::ProgramResp(resp) =
-                stream.next().await.unwrap().unwrap().response.unwrap()
-            else {
-                panic!()
-            };
-            if !apply_program_resp_to_builder(
-                &QueryBuilderConfig::default(),
-                &mut builder,
-                resp,
-                |_, _| (),
-            )
-            .unwrap()
-            {
-                break;
-            }
-            num_resp += 1;
-        }
-
-        assert_eq!(num_resp, 3);
-        assert_debug_snapshot!(builder.into_ret());
-    }
-
     #[tokio::test]
     async fn request_interupted() {
         let tmp = tempdir().unwrap();
@@ -615,6 +562,9 @@ pub mod test {
             move || fsm_builder_driver(&trace, builder)
         });
 
-        (ReceiverStream::new(receiver), ValidateTraceBuilder::new(trace))
+        (
+            ReceiverStream::new(receiver),
+            ValidateTraceBuilder::new(trace),
+        )
     }
 }