/// Execute a transaction against QLDB, retrying as necessary.
///
/// This function is the primary way you should interact with QLDB. The
/// driver will acquire a connection and open a [`Transaction`], handing it
/// to your code (the closure you pass in). The driver will run your
/// transaction and attempt to commit it. While executing the transaction,
/// failures may occur. The driver will retry (acquire a new transaction and
/// run your code again). This means your code *must be idempotent*.
///
/// If you wish to get results out of your transaction, you must return them
/// from your closure. Do not attempt to write to variables outside of your
/// closure! You should consider code running inside the closure as seeing
/// speculative results that are only confirmed once the transaction
/// commits.
Execute a transaction against QLDB, retrying as necessary.
This function is the primary way you should interact with QLDB. The
driver will acquire a connection and open a [`Transaction`], handing it
to your code (the closure you pass in). The driver will run your
transaction and attempt to commit it. While executing the transaction,
failures may occur. The driver will retry (acquire a new transaction and
run your code again). This means your code *must be idempotent*.

If you wish to get results out of your transaction, you must return them
from your closure. Do not attempt to write to variables outside of your
closure. You should consider code running inside the closure as seeing
speculative results that are only confirmed once the transaction
commits.
Execute a transaction against QLDB retrying as necessary . This function is the primary way you should interact with QLDB . The driver will acquire a connection and open a [ ` Transaction ` ] handing it to your code ( the closure you pass in ) . The driver will run your transaction and attempt to commit it . While executing the transaction failures may occur . The driver will retry ( acquire a new transaction and run your code again ) . This means your code * must be idempotent * . If you wish to get results out of your transaction you must return them from your closure . Do not attempt to write to variables outside of your closure . You should consider code running inside the closure as seeing speculative results that are only confirmed once the transaction commits .
pub async fn transact<F, Fut, R>(&self, transaction: F) -> Result<R>
    where
        Fut: Future<Output = Result<TransactionAttemptResult<R>>>,
        F: Fn(TransactionAttempt<C>) -> Fut,
    {
        let mut attempt_number = 0u32;
        loop {
            attempt_number += 1;
            let pooled_session = self.session_pool.get().await?;
            let tx = match TransactionAttempt::start(pooled_session.clone()).await {
                Ok(tx) => tx,
                Err(e) => {
                    // FIXME: Include some sort of sleep and attempt cap.
                    debug!(error = %e, "unable to start a session, will retry");
                    continue;
                }
            };
            let tx_id = tx.id.clone();
            // Run the user's transaction. They can run methods on
            // [`Transaction`] such as [`execute_statement`]. When this future
            // completes, one of 4 things could have happened:
            //
            // 1. The user code ended with `tx.commit(R)`. This means we get
            // instructions back to attempt a commit. If the commit succeeds,
            // the user gets their data and we're done.
            //
            // 2. The user code ended with `tx.abort(R)`. We're done, regardless
            // of whether the API request to `abort` succeeded.
            //
            // 3. The user code returned an `Err`:
            //    a. If it is a QldbError that is retriable (e.g. consider a
            //    communications failure), then the transaction is retried.
            //    b. Otherwise, we return the error to the user. Arbitrary
            //    application types are *not* retried.
            let qldb_err = match transaction(tx).await {
                Ok(TransactionAttemptResult::Committed { user_data, .. }) => return Ok(user_data),
                Ok(TransactionAttemptResult::Aborted) => Err(TransactionError::Aborted)?,
                Err(e) => {
                    debug!(
                        error = %e,
                        id = &tx_id[..],
                        "transaction failed with error"
                    );
                    match e.downcast::<QldbError>() {
                        Ok(qldb_err) => qldb_err,
                        // This branch means the transaction block failed with a
                        // non-QldbError. This means something unrelated to QLDB
                        // went wrong. We don't retry on arbitrary user errors.
                        Err(other) => {
                            return Err(other);
                        }
                    }
                }
            };
            let retry_ins = {
                let policy = self.transaction_retry_policy.lock().await;
                policy.on_err(&qldb_err, attempt_number)
            };
            if retry_ins.should_retry {
                debug!(
                    error = %qldb_err,
                    attempt_number, "Error comitting, will retry"
                );
                if let Some(duration) = retry_ins.delay {
                    debug!(
                        delay = duration.as_millis().to_string().as_str(),
                        "Retry will be delayed"
                    );
                    sleep(duration).await;
                }
            } else {
                debug!(
                    "Not retrying after {} attempts due to error: {}",
                    attempt_number, qldb_err
                );
                return Err(TransactionError::MaxAttempts {
                    last_err: qldb_err,
                    attempts: attempt_number,
                })?;
            }
            // Here we retry. We're in a loop, remember!
        }
    }
pub async fn transact < F , Fut , R > ( & self , transaction : F ) -> Result < R > where Fut : Future < Output = Result < TransactionAttemptResult < R > > > , F : Fn ( TransactionAttempt < C > ) -> Fut , { let mut attempt_number = 0u32 ; loop { attempt_number += 1 ; let pooled_session = self . session_pool . get ( ) . await ? ; let tx = match TransactionAttempt :: start ( pooled_session . clone ( ) ) . await { Ok ( tx ) => tx , Err ( e ) => { // FIXME: Include some sort of sleep and attempt cap. debug ! ( error e " " ) ; continue ; } } ; let tx_id = tx . id . clone ( ) ; // Run the user's transaction. They can run methods on // [`Transaction`] such as [`execute_statement`]. When this future // completes, one of 4 things could have happened: // // 1. The user code ended with `tx.commit(R)`. This means we get // instructions back to attempt a commit. If the commit succeeds, // the user gets their data and we're done. // // 2. The user code ended with `tx.abort(R)`. We're done, regardless // of whether the API request to `abort` succeeded. // // 3. The user code returned an `Err`: //    a. If it is a QldbError that is retriable (e.g. consider a //    communications failure), then the transaction is retried. //    b. Otherwise, we return the error to the user. Arbitrary //    application types are *not* retried. let qldb_err = match transaction ( tx ) . await { Ok ( TransactionAttemptResult :: Committed { user_data , .. } ) => return Ok ( user_data ) , Ok ( TransactionAttemptResult :: Aborted ) => Err ( TransactionError :: Aborted ) ? , Err ( e ) => { debug ! ( error e id tx_id [ ] " " ) ; match e . downcast :: < QldbError > ( ) { Ok ( qldb_err ) => qldb_err , // This branch means the transaction block failed with a // non-QldbError. This means something unrelated to QLDB // went wrong. We don't retry on arbitrary user errors. Err ( other ) => { return Err ( other ) ; } } } } ; let retry_ins = { let policy = self . transaction_retry_policy . lock ( ) . await ; policy . on_err ( & qldb_err , attempt_number ) } ; if retry_ins . should_retry { debug ! ( error qldb_err attempt_number " " ) ; if let Some ( duration ) = retry_ins . delay { debug ! ( delay duration as_millis ( ) to_string ( ) as_str ( ) " " ) ; sleep ( duration ) . await ; } } else { debug ! ( " " attempt_number qldb_err ) ; return Err ( TransactionError :: MaxAttempts { last_err : qldb_err , attempts : attempt_number , } ) ? ; } // Here we retry. We're in a loop, remember! } }
Execute a transaction against QLDB, retrying as necessary.
Execute a transaction against QLDB retrying as necessary .
&self
transaction
F