fmt

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::*;
@@ -615,6 +617,9 @@ pub mod test {
             move || fsm_builder_driver(&trace, builder)
         });
 
-        (ReceiverStream::new(receiver), ValidateTraceBuilder::new(trace))
+        (
+            ReceiverStream::new(receiver),
+            ValidateTraceBuilder::new(trace),
+        )
     }
 }