Erase SmithyConnector

This lets us move the PooledConnection into TransactionAttempt. Having
that type have a lifetime is really painful.

At this point, the doctest passes, but it's ugly!

Author: marcbowes

amazon-qldb-driver-core/src/driver.rs

@@ -1,4 +1,3 @@
-use aws_hyper::{DynConnector, SmithyConnector};
 use aws_sdk_qldbsessionv2::{Client, Config};
 use bb8::Pool;
 use std::sync::Arc;
@@ -64,7 +63,7 @@ impl QldbDriverBuilder {
     pub async fn sdk_config(
         self,
         sdk_config: &aws_types::config::Config,
-    ) -> Result<QldbDriver<DynConnector>, BuilderError> {
+    ) -> Result<QldbDriver, BuilderError> {
         let client = Client::new(sdk_config);
         Ok(self.build_with_client(client).await?)
     }
@@ -73,18 +72,12 @@ impl QldbDriverBuilder {
     ///
     /// Note that `config` is the service-specific (QldbSession) config. For
     /// shared config, see [`sdk_config`].
-    pub async fn config(self, config: Config) -> Result<QldbDriver<DynConnector>, BuilderError> {
+    pub async fn config(self, config: Config) -> Result<QldbDriver, BuilderError> {
         let client = Client::from_conf(config);
         Ok(self.build_with_client(client).await?)
     }
 
-    pub async fn build_with_client<C>(
-        self,
-        client: Client<C>,
-    ) -> Result<QldbDriver<C>, BuilderError>
-    where
-        C: SmithyConnector,
-    {
+    pub async fn build_with_client(self, client: Client) -> Result<QldbDriver, BuilderError> {
         let ledger_name = self.ledger_name.ok_or(BuilderError::UsageError(format!(
             "ledger_name must be initialized"
         )))?;
@@ -148,19 +141,13 @@ impl QldbDriverBuilder {
 /// we wrap all policies in a Mutex. Performance of a Mutex is unlikely to be
 /// the limiting factor in a QLDB application.
 #[derive(Clone)]
-pub struct QldbDriver<C>
-where
-    C: SmithyConnector,
-{
+pub struct QldbDriver {
     pub ledger_name: Arc<String>,
-    session_pool: Arc<Pool<QldbSessionV2Manager<C>>>,
+    session_pool: Arc<Pool<QldbSessionV2Manager>>,
     transaction_retry_policy: Arc<Mutex<Box<dyn TransactionRetryPolicy + Send + Sync>>>,
 }
 
-impl<C> QldbDriver<C>
-where
-    C: SmithyConnector,
-{
+impl QldbDriver {
     pub fn ledger_name(&self) -> String {
         (*self.ledger_name).clone()
     }
@@ -190,15 +177,17 @@ where
         loop {
             attempt_number += 1;
 
-            // Connection failures at this point have already been retried by
-            // the pool, and so we give up at this point.
-            let mut pooled_session =
-                self.session_pool
-                    .get()
-                    .await
-                    .map_err(|bb8| TransactError::CommunicationError {
-                        source: Box::new(bb8),
-                    })?;
+            // We used `get_owned` to hide the lifetime of the pool from
+            // customers. This is important because otherwise the future
+            // inherits this lifetime and things get messy (shows up as:
+            // `TransactionAttempt<'pool>`).
+            let pooled_session = self.session_pool.get_owned().await.map_err(|bb8| {
+                // Connection failures at this point have already been retried by
+                // the pool, and so we give up at this point.
+                TransactError::CommunicationError {
+                    source: Box::new(bb8),
+                }
+            })?;
 
             // Note that `PooledConnection` uses the `Drop` trait to hand the
             // connection back to the pool. It'd be really nice to hand the
@@ -210,7 +199,7 @@ where
             // Rust will guarantee we do it at some point), but putting the
             // connection back in the pool before sleeping will probably be
             // beneficial.
-            let tx = match TransactionAttempt::start(&mut pooled_session).await {
+            let tx = match TransactionAttempt::start(pooled_session).await {
                 Ok(tx) => tx,
                 Err(e) => {
                     // FIXME: Include some sort of sleep and attempt cap.
@@ -258,10 +247,6 @@ where
                 }
             };
 
-            // See above note. We put the connection back in the pool before
-            // adding any delay.
-            drop(pooled_session);
-
             let retry_ins = {
                 let policy = self.transaction_retry_policy.lock().await;
                 policy.on_err(&comm_err, attempt_number)

amazon-qldb-driver-core/src/error.rs

@@ -70,6 +70,7 @@ impl From<aws_sdk_qldbsessionv2::model::TransactionError> for GenericError {
 /// due to user code (e.g. deserializing a QLDB document into a local type), or
 /// from interacting with the QLDB service.
 #[derive(Debug, Error)]
+#[non_exhaustive]
 pub enum TransactError<E>
 where
     E: std::error::Error + 'static,
@@ -94,7 +95,7 @@ where
         source: BoxError,
     },
 
-    #[error("transaction was aborted")]
+    #[error("transaction was aborted by user code")]
     Aborted,
 
     #[error("transaction failed after {attempts} attempts, last error: {last_err}")]

amazon-qldb-driver-core/src/pool.rs

@@ -1,5 +1,5 @@
 use async_trait::async_trait;
-use aws_hyper::{SdkError, SmithyConnector};
+use aws_hyper::{DynConnector, SdkError, SmithyConnector};
 use aws_sdk_qldbsessionv2::{
     error::SendCommandError,
     model::{CommandStream, ResultStream},
@@ -39,10 +39,7 @@ impl ErrorSink<ConnectionError> for QldbErrorLoggingErrorSink {
 
 // FIXME: This is not generic over the client (it uses the rust awssdk
 // directly), which means we cannot change the sdk (e.g. to support javascript).
-pub struct QldbSessionV2Manager<C>
-where
-    C: SmithyConnector,
-{
+pub struct QldbSessionV2Manager<C = DynConnector> {
     client: Client<C>,
     pub ledger_name: String,
 }

amazon-qldb-driver-core/src/transaction.rs

@@ -2,6 +2,7 @@ use aws_sdk_qldbsessionv2::model::{
     AbortTransactionRequest, CommandStream, CommitTransactionRequest, ExecuteStatementRequest,
     FetchPageRequest, ResultStream, StartTransactionRequest,
 };
+use bb8::PooledConnection;
 use bytes::Bytes;
 use ion_c_sys::reader::IonCReaderHandle;
 use ion_c_sys::result::IonCError;
@@ -10,7 +11,7 @@ use std::marker::PhantomData;
 use tracing::debug;
 
 use crate::error;
-use crate::pool::QldbHttp2Connection;
+use crate::pool::QldbSessionV2Manager;
 use crate::{error::TransactError, execution_stats::ExecutionStats};
 
 /// The results of executing a statement.
@@ -68,9 +69,9 @@ pub enum TransactionDisposition<R> {
 /// succeeds!
 ///
 /// `E` represents any custom error variant the user may throw.
-pub struct TransactionAttempt<'pool, E> {
-    /// A pool connection that we'll send our commands down.
-    connection: &'pool mut QldbHttp2Connection,
+pub struct TransactionAttempt<E> {
+    /// A pooled connection that we'll send our commands down.
+    connection: PooledConnection<'static, QldbSessionV2Manager>,
 
     /// The id of this transaction attempt. This is a speculative transaction
     /// id. That is, if the transaction commits, then this id is the id of the
@@ -93,13 +94,13 @@ pub struct TransactionAttempt<'pool, E> {
     err: PhantomData<E>,
 }
 
-impl<'pool, E> TransactionAttempt<'pool, E>
+impl<E> TransactionAttempt<E>
 where
     E: std::error::Error + 'static,
 {
     pub(crate) async fn start(
-        connection: &'pool mut QldbHttp2Connection,
-    ) -> Result<TransactionAttempt<'pool, E>, TransactError<E>> {
+        mut connection: PooledConnection<'static, QldbSessionV2Manager>,
+    ) -> Result<TransactionAttempt<E>, TransactError<E>> {
         let mut accumulated_execution_stats = ExecutionStats::default();
         let resp = connection
             .send_streaming_command(CommandStream::StartTransaction(
@@ -127,7 +128,7 @@ where
         })
     }
 
-    pub fn statement<S>(&mut self, statement: S) -> StatementBuilder<'pool, '_, E>
+    pub fn statement<S>(&mut self, statement: S) -> StatementBuilder<'_, E>
     where
         S: Into<String>,
     {
@@ -287,19 +288,16 @@ where
     }
 }
 
-pub struct StatementBuilder<'pool, 'tx, E> {
-    attempt: &'tx mut TransactionAttempt<'pool, E>,
+pub struct StatementBuilder<'tx, E> {
+    attempt: &'tx mut TransactionAttempt<E>,
     statement: Statement,
 }
 
-impl<'pool, 'tx, E> StatementBuilder<'pool, 'tx, E>
+impl<'tx, E> StatementBuilder<'tx, E>
 where
     E: std::error::Error + 'static,
 {
-    fn new(
-        attempt: &'tx mut TransactionAttempt<'pool, E>,
-        partiql: String,
-    ) -> StatementBuilder<'pool, 'tx, E> {
+    fn new(attempt: &'tx mut TransactionAttempt<E>, partiql: String) -> StatementBuilder<'tx, E> {
         StatementBuilder {
             attempt,
             statement: Statement {
@@ -313,7 +311,7 @@ where
     // 1. need an IonElement so we can hash it. in the future, we hope to remove this as a requirement
     // 2. perhaps we want an in-crate trait for coherency reasons
     // TODO: make public when ready
-    pub fn param<B>(mut self, param: B) -> StatementBuilder<'pool, 'tx, E>
+    pub fn param<B>(mut self, param: B) -> StatementBuilder<'tx, E>
     where
         B: Into<Bytes>,
     {

amazon-qldb-driver/src/lib.rs

@@ -2,7 +2,7 @@
 //!
 //! Usage example:
 //!
-//! ```
+//! ```no_run
 //! use amazon_qldb_driver::awssdk::Config;
 //! use amazon_qldb_driver::{QldbDriverBuilder, TransactError};
 //! use tokio;