Make install extensions fn async

compute_tools/src/compute.rs

@@ -1368,22 +1368,28 @@ LIMIT 100",
         download_size
     }
 
-    pub fn install_extension(
+    pub async fn install_extension(
         &self,
         ext_name: &str,
         db_name: &str,
         ext_version: &str,
     ) -> Result<String> {
+        use tokio_postgres::config::Config;
+        use tokio_postgres::NoTls;
+
         let mut conf = Config::from_str(self.connstr.as_str()).unwrap();
         conf.dbname(db_name);
 
-        let mut db_client = conf
+        let (db_client, conn) = conf
             .connect(NoTls)
+            .await
             .context("Failed to connect to the database")?;
+        tokio::spawn(conn);
 
         let version_query = "SELECT extversion FROM pg_extension WHERE extname = $1";
         let version: Result<String> = db_client
             .query_one(version_query, &[&ext_name])
+            .await
             .context(format!("Failed to execute query: {}", version_query))
             .map(|row| row.get(0));
 
@@ -1391,14 +1397,17 @@ LIMIT 100",
             if installed_version == ext_version {
                 return Ok(installed_version);
             }
-            let query = "ALTER EXTENSION $1 UPDATE TO $2";
+            let query = format!("ALTER EXTENSION {ext_name:?} UPDATE TO {ext_version:?}");
             db_client
-                .execute(query, &[&ext_name, &ext_version])
+                .simple_query(&query)
+                .await
                 .context(format!("Failed to execute query: {}", query))?;
         } else {
-            let query = "CREATE EXTENSION IF NOT EXISTS $1 WITH VERSION $2";
+            let query =
+                format!("CREATE EXTENSION IF NOT EXISTS {ext_name:?} WITH VERSION {ext_version:?}");
             db_client
-                .execute(query, &[&ext_name, &ext_version])
+                .simple_query(&query)
+                .await
                 .context(format!("Failed to execute query: {}", query))?;
         }
 

compute_tools/src/http/api.rs

@@ -114,13 +114,14 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
 
             let request = hyper::body::to_bytes(req.into_body()).await.unwrap();
             let request = serde_json::from_slice::<ExtensionInstallRequest>(&request).unwrap();
-            let res =
-                compute.install_extension(&request.extension, &request.database, &request.version);
+            let res = compute
+                .install_extension(&request.extension, &request.database, &request.version)
+                .await;
             match res {
-                Ok(res) => render_json(Body::from(
+                Ok(version) => render_json(Body::from(
                     serde_json::to_string(&ExtensionInstallResult {
                         extension: request.extension,
-                        version: res,
+                        version,
                     })
                     .unwrap(),
                 )),