PVC-705: switch to typed queries for performance benefits

Currently, each query requires three RTTs to the DB. By switching to typed queries, we
can do the parse, bind and execute all in a single RTT.

Queries which use `SqlNamespaceStatus` were not ported due to me not being able to figure
out which type they have. It's a user-generated type, and I'm not sure there's a matching
type in `postgres-types::Type`. In any case, this just means we eat the extra RTTs when
creating or updating namespaces. This feels fine to me, since at least the hot path is
able to be typed.

Related tokio-postgres PR: sfackler/rust-postgres#1147

Author: fisherdarling

crates/plexi-backend-postgres/src/schema/signature.rs

@@ -1,7 +1,8 @@
+use anyhow::Context;
 use chrono::{DateTime, Utc};
 use const_format::formatc;
 use plexi_core::{Ciphersuite, Epoch, SignatureResponse};
-use postgres_types::{FromSql, ToSql};
+use postgres_types::{FromSql, ToSql, Type};
 use tokio_postgres::{Row, Transaction};
 
 use crate::schema::namespace::{SqlAccountId, SqlCiphersuite};
@@ -49,21 +50,21 @@ impl SqlSignature {
         tracing::info!("getting signature by epoch");
 
         let row = tx
-            .query_opt(
+            .query_typed(
                 formatc!(
                     "SELECT {SIGNATURE_FIELDS}
                     FROM signature 
                     WHERE account_id=$1 AND namespace=$2 AND epoch=$3",
                 ),
                 &[
-                    &SqlAccountId::from(account_id),
-                    &namespace,
-                    &SqlEpoch::from(epoch),
+                    (&SqlAccountId::from(account_id), Type::INT8),
+                    (&namespace, Type::TEXT),
+                    (&SqlEpoch::from(epoch), Type::INT8),
                 ],
             )
             .await?;
 
-        row.map(SqlSignature::try_from).transpose()
+        row.first().cloned().map(SqlSignature::try_from).transpose()
     }
 
     pub async fn get_by_digest(
@@ -75,17 +76,21 @@ impl SqlSignature {
         tracing::info!("getting signature by digest");
 
         let row = tx
-            .query_opt(
+            .query_typed(
                 formatc!(
                     "SELECT {SIGNATURE_FIELDS}
                     FROM signature 
                     WHERE account_id=$1 AND namespace=$2 AND digest=$3",
                 ),
-                &[&SqlAccountId::from(account_id), &namespace, &digest],
+                &[
+                    (&SqlAccountId::from(account_id), Type::INT8),
+                    (&namespace, Type::TEXT),
+                    (&digest, Type::BYTEA),
+                ],
             )
             .await?;
 
-        row.map(SqlSignature::try_from).transpose()
+        row.first().cloned().map(SqlSignature::try_from).transpose()
     }
 
     /// In this flow, we assume that the transaction has serialized isolation level.
@@ -98,27 +103,30 @@ impl SqlSignature {
         tracing::info!("inserting signature");
 
         let row = tx
-            .query_one(
+            .query_typed(
                 formatc!(
                     "INSERT INTO signature VALUES 
                     ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                     RETURNING {SIGNATURE_FIELDS}",
                 ),
                 &[
-                    &SqlAccountId::from(account_id),                         // $1
-                    &sig_resp.namespace(),                                   // $2
-                    &SqlEpoch::from(*sig_resp.epoch()),                      // $3
-                    &ts_to_datetime(sig_resp.timestamp())?,                  //$4
-                    &sig_resp.digest().as_slice(),                           // $5
-                    &SqlCiphersuite::from(*sig_resp.ciphersuite()),          // $6
-                    &sig_resp.signature().as_slice(),                        // $7
-                    &sig_resp.key_id().map(|v| vec![v]).unwrap_or_default(), // $8
-                    &sig_resp.serialized_message(),                          // $9
+                    (&SqlAccountId::from(account_id), Type::INT8),    // $1
+                    (&sig_resp.namespace(), Type::TEXT),              // $2
+                    (&SqlEpoch::from(*sig_resp.epoch()), Type::INT8), // $3
+                    (&ts_to_datetime(sig_resp.timestamp())?, Type::TIMESTAMPTZ), //$4
+                    (&sig_resp.digest().as_slice(), Type::BYTEA),     // $5
+                    (&SqlCiphersuite::from(*sig_resp.ciphersuite()), Type::INT4), // $6
+                    (&sig_resp.signature().as_slice(), Type::BYTEA),  // $7
+                    (
+                        &sig_resp.key_id().map(|v| vec![v]).unwrap_or_default(),
+                        Type::BYTEA,
+                    ), // $8
+                    (&sig_resp.serialized_message(), Type::BYTEA),    // $9
                 ],
             )
             .await?;
 
-        SqlSignature::try_from(row)
+        SqlSignature::try_from(row.first().context("empty insert")?.clone())
     }
 }
 

crates/plexi-backend-postgres/src/schema/signing_key.rs

@@ -4,6 +4,7 @@ use const_format::formatc;
 use plexi_core::auditor::KeyInfo;
 use plexi_core::crypto::ed25519_dalek::{SecretKey, SigningKey};
 use plexi_core::crypto::KeyPair;
+use postgres_types::Type;
 use tokio_postgres::{Row, Transaction};
 
 #[derive(Debug, Clone, Copy)]
@@ -45,7 +46,7 @@ impl SqlSigningKey {
         tracing::info!("getting first available signing key");
 
         let row = tx
-            .query_opt(
+            .query_typed(
                 formatc!(
                     "SELECT {SIGNING_KEY_FIELDS}
                     FROM signing_key 
@@ -57,7 +58,9 @@ impl SqlSigningKey {
             )
             .await?;
 
-        row.map(SqlSigningKey::try_from)
+        row.first()
+            .cloned()
+            .map(SqlSigningKey::try_from)
             .transpose()?
             .context("auditor not initialized")
     }
@@ -66,7 +69,7 @@ impl SqlSigningKey {
         tracing::info!("getting all signing keys");
 
         let row = tx
-            .query(
+            .query_typed(
                 formatc!(
                     "SELECT {SIGNING_KEY_FIELDS}
                     FROM signing_key 
@@ -84,25 +87,25 @@ impl SqlSigningKey {
 
     pub async fn insert(&self, tx: &mut Transaction<'_>) -> anyhow::Result<SqlSigningKey> {
         let row = tx
-            .query_one(
+            .query_typed(
                 formatc!(
                     "INSERT INTO signing_key VALUES 
                     ($1, $2, $3, $4, $5, $6, $7)
                     RETURNING {SIGNING_KEY_FIELDS}"
                 ),
                 &[
-                    &self.verifying_key,  // $1
-                    &self.signing_key,    // $2
-                    &(self.curve as i32), // $3
-                    &self.not_before,     // $4
-                    &self.created_at,     // $5
-                    &self.updated_at,     // $6
-                    &self.disabled_at,    // $7
+                    (&self.verifying_key, Type::BYTEA),     // $1
+                    (&self.signing_key, Type::BYTEA),       // $2
+                    (&(self.curve as i32), Type::INT4),     // $3
+                    (&self.not_before, Type::TIMESTAMPTZ),  // $4
+                    (&self.created_at, Type::TIMESTAMPTZ),  // $5
+                    (&self.updated_at, Type::TIMESTAMPTZ),  // $6
+                    (&self.disabled_at, Type::TIMESTAMPTZ), // $7
                 ],
             )
             .await?;
 
-        SqlSigningKey::try_from(row)
+        SqlSigningKey::try_from(row.first().context("empty insert")?.clone())
     }
 }