use tokio-postgres in outbound-pg (#779)

* use `tokio-postgres` in `outbound-pg`

Since `postgres` just wraps the async `tokio-postgres` library in a synchronous
wrapper, and `outbound-pg` exports async methods, it makes more sense to use
`tokio-postgres` directly.

Addresses #776.

Signed-off-by: Joel Dice <joel.dice@fermyon.com>

* fix PgError mapping

Signed-off-by: Joel Dice <joel.dice@fermyon.com>

Signed-off-by: Joel Dice <joel.dice@fermyon.com>

Author: dicej

Cargo.lock

@@ -8,7 +8,7 @@ doctest = false
 
 [dependencies]
 anyhow = "1.0"
-postgres = { version = "0.19.3" }
+tokio-postgres = { version = "0.7.7" }
 spin-engine = { path = "../engine" }
 spin-manifest = { path = "../manifest" }
 tokio = { version = "1", features = [ "rt-multi-thread" ] }
@@ -18,4 +18,3 @@ tracing = { version = "0.1", features = [ "log" ] }
 git = "https://github.com/bytecodealliance/wit-bindgen"
 rev = "cb871cfa1ee460b51eb1d144b175b9aab9c50aba"
 features = ["async"]
-

crates/outbound-pg/Cargo.toml

@@ -1,6 +1,10 @@
 use anyhow::anyhow;
 use outbound_pg::*;
-use postgres::{types::ToSql, types::Type, Client, NoTls, Row};
+use tokio_postgres::{
+    tls::NoTlsStream,
+    types::{ToSql, Type},
+    Connection, NoTls, Row, Socket,
+};
 
 pub use outbound_pg::add_to_linker;
 use spin_engine::{
@@ -41,23 +45,24 @@ impl outbound_pg::OutboundPg for OutboundPg {
         statement: &str,
         params: Vec<ParameterValue<'_>>,
     ) -> Result<u64, PgError> {
-        // TODO: consider using async tokio-postgres crate
-        tokio::task::block_in_place(|| {
-            let mut client = Client::connect(address, NoTls)
-                .map_err(|e| PgError::ConnectionFailed(format!("{:?}", e)))?;
-
-            let params: Vec<&(dyn ToSql + Sync)> = params
-                .iter()
-                .map(to_sql_parameter)
-                .collect::<anyhow::Result<Vec<_>>>()
-                .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
-
-            let nrow = client
-                .execute(statement, params.as_slice())
-                .map_err(|e| PgError::ValueConversionFailed(format!("{:?}", e)))?;
-
-            Ok(nrow)
-        })
+        let (client, connection) = tokio_postgres::connect(address, NoTls)
+            .await
+            .map_err(|e| PgError::ConnectionFailed(format!("{:?}", e)))?;
+
+        spawn(connection);
+
+        let params: Vec<&(dyn ToSql + Sync)> = params
+            .iter()
+            .map(to_sql_parameter)
+            .collect::<anyhow::Result<Vec<_>>>()
+            .map_err(|e| PgError::ValueConversionFailed(format!("{:?}", e)))?;
+
+        let nrow = client
+            .execute(statement, params.as_slice())
+            .await
+            .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
+
+        Ok(nrow)
     }
 
     async fn query(
@@ -66,36 +71,38 @@ impl outbound_pg::OutboundPg for OutboundPg {
         statement: &str,
         params: Vec<ParameterValue<'_>>,
     ) -> Result<RowSet, PgError> {
-        tokio::task::block_in_place(|| {
-            let mut client = Client::connect(address, NoTls)
-                .map_err(|e| PgError::ConnectionFailed(format!("{:?}", e)))?;
-
-            let params: Vec<&(dyn ToSql + Sync)> = params
-                .iter()
-                .map(to_sql_parameter)
-                .collect::<anyhow::Result<Vec<_>>>()
-                .map_err(|e| PgError::BadParameter(format!("{:?}", e)))?;
-
-            let results = client
-                .query(statement, params.as_slice())
-                .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
-
-            if results.is_empty() {
-                return Ok(RowSet {
-                    columns: vec![],
-                    rows: vec![],
-                });
-            }
+        let (client, connection) = tokio_postgres::connect(address, NoTls)
+            .await
+            .map_err(|e| PgError::ConnectionFailed(format!("{:?}", e)))?;
+
+        spawn(connection);
+
+        let params: Vec<&(dyn ToSql + Sync)> = params
+            .iter()
+            .map(to_sql_parameter)
+            .collect::<anyhow::Result<Vec<_>>>()
+            .map_err(|e| PgError::BadParameter(format!("{:?}", e)))?;
+
+        let results = client
+            .query(statement, params.as_slice())
+            .await
+            .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
+
+        if results.is_empty() {
+            return Ok(RowSet {
+                columns: vec![],
+                rows: vec![],
+            });
+        }
 
-            let columns = infer_columns(&results[0]);
-            let rows = results
-                .iter()
-                .map(convert_row)
-                .collect::<Result<Vec<_>, _>>()
-                .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
+        let columns = infer_columns(&results[0]);
+        let rows = results
+            .iter()
+            .map(convert_row)
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(|e| PgError::QueryFailed(format!("{:?}", e)))?;
 
-            Ok(RowSet { columns, rows })
-        })
+        Ok(RowSet { columns, rows })
     }
 }
 
@@ -110,10 +117,10 @@ fn to_sql_parameter<'a>(value: &'a ParameterValue) -> anyhow::Result<&'a (dyn To
         ParameterValue::Int16(v) => Ok(v),
         ParameterValue::Floating32(v) => Ok(v),
         ParameterValue::Floating64(v) => Ok(v),
-        ParameterValue::Uint8(_) => Err(anyhow!("Postgres does not support unsigned integers")),
-        ParameterValue::Uint16(_) => Err(anyhow!("Postgres does not support unsigned integers")),
-        ParameterValue::Uint32(_) => Err(anyhow!("Postgres does not support unsigned integers")),
-        ParameterValue::Uint64(_) => Err(anyhow!("Postgres does not support unsigned integers")),
+        ParameterValue::Uint8(_)
+        | ParameterValue::Uint16(_)
+        | ParameterValue::Uint32(_)
+        | ParameterValue::Uint64(_) => Err(anyhow!("Postgres does not support unsigned integers")),
         ParameterValue::Str(v) => Ok(v),
         ParameterValue::Binary(v) => Ok(v),
         ParameterValue::DbNull => Ok(&DB_NULL),
@@ -153,15 +160,15 @@ fn convert_data_type(pg_type: &Type) -> DbDataType {
     }
 }
 
-fn convert_row(row: &Row) -> Result<Vec<DbValue>, postgres::Error> {
+fn convert_row(row: &Row) -> Result<Vec<DbValue>, tokio_postgres::Error> {
     let mut result = Vec::with_capacity(row.len());
     for index in 0..row.len() {
         result.push(convert_entry(row, index)?);
     }
     Ok(result)
 }
 
-fn convert_entry(row: &Row, index: usize) -> Result<DbValue, postgres::Error> {
+fn convert_entry(row: &Row, index: usize) -> Result<DbValue, tokio_postgres::Error> {
     let column = &row.columns()[index];
     let value = match column.type_() {
         &Type::BOOL => {
@@ -238,3 +245,11 @@ fn convert_entry(row: &Row, index: usize) -> Result<DbValue, postgres::Error> {
     };
     Ok(value)
 }
+
+fn spawn(connection: Connection<Socket, NoTlsStream>) {
+    tokio::spawn(async move {
+        if let Err(e) = connection.await {
+            tracing::warn!("Postgres connection error: {}", e);
+        }
+    });
+}