Struct mongodb::ClientSession

pub struct ClientSession { /* private fields */ }

A MongoDB client session. This struct represents a logical session used for ordering sequential operations. To create a ClientSession, call start_session on a Client.

ClientSession instances are not thread safe or fork safe. They can only be used by one thread or process at a time.

Transactions

Transactions are used to execute a series of operations across multiple documents and collections atomically. For more information about when and how to use transactions in MongoDB, see the manual.

Replica set transactions are supported on MongoDB 4.0+. Sharded transactions are supported on MongoDDB 4.2+. Transactions are associated with a ClientSession. To begin a transaction, call ClientSession::start_transaction on a ClientSession. The ClientSession must be passed to operations to be executed within the transaction.

use mongodb::{
    bson::{doc, Document},
    error::{Result, TRANSIENT_TRANSACTION_ERROR, UNKNOWN_TRANSACTION_COMMIT_RESULT},
    options::{Acknowledgment, ReadConcern, TransactionOptions, WriteConcern},
    ClientSession,
    Collection,
};

let mut session = client.start_session(None).await?;
let options = TransactionOptions::builder()
    .read_concern(ReadConcern::majority())
    .write_concern(WriteConcern::builder().w(Acknowledgment::Majority).build())
    .build();
session.start_transaction(options).await?;
// A "TransientTransactionError" label indicates that the entire transaction can be retried
// with a reasonable expectation that it will succeed.
while let Err(error) = execute_transaction(&coll, &mut session).await {
    if !error.contains_label(TRANSIENT_TRANSACTION_ERROR) {
        break;
    }
}

async fn execute_transaction(coll: &Collection<Document>, session: &mut ClientSession) -> Result<()> {
    coll.insert_one_with_session(doc! { "x": 1 }, None, session).await?;
    coll.delete_one_with_session(doc! { "y": 2 }, None, session).await?;
    // An "UnknownTransactionCommitResult" label indicates that it is unknown whether the
    // commit has satisfied the write concern associated with the transaction. If an error
    // with this label is returned, it is safe to retry the commit until the write concern is
    // satisfied or an error without the label is returned.
    loop {
        let result = session.commit_transaction().await;
        if let Err(ref error) = result {
            if error.contains_label(UNKNOWN_TRANSACTION_COMMIT_RESULT) {
                continue;
            }
        }
        result?
    }
}

Implementations

impl ClientSession

pub fn client(&self) -> Client

The client used to create this session.

pub fn id(&self) -> &Document

The id of this session.

pub fn cluster_time(&self) -> Option<&ClusterTime>

The highest seen cluster time this session has seen so far. This will be None if this session has not been used in an operation yet.

pub fn options(&self) -> Option<&SessionOptions>

The options used to create this session.

pub fn advance_cluster_time(&mut self, to: &ClusterTime)

Set the cluster time to the provided one if it is greater than this session’s highest seen cluster time or if this session’s cluster time is None.

pub fn advance_operation_time(&mut self, ts: Timestamp)

Advance operation time for this session. If the provided timestamp is earlier than this session’s current operation time, then the operation time is unchanged.

pub fn operation_time(&self) -> Option<Timestamp>

The operation time returned by the last operation executed in this session.

pub async fn start_transaction( &mut self, options: impl Into<Option<TransactionOptions>> ) -> Result<()>

Starts a new transaction on this session with the given TransactionOptions. If no options are provided, the session’s defaultTransactionOptions will be used. This session must be passed into each operation within the transaction; otherwise, the operation will be executed outside of the transaction.

Errors returned from operations executed within a transaction may include a crate::error::TRANSIENT_TRANSACTION_ERROR label. This label indicates that the entire transaction can be retried with a reasonable expectation that it will succeed.

Transactions are supported on MongoDB 4.0+. The Rust driver currently only supports transactions on replica sets.

session.start_transaction(None).await?;
let result = coll.insert_one_with_session(doc! { "x": 1 }, None, &mut session).await?;
session.commit_transaction().await?;

pub async fn commit_transaction(&mut self) -> Result<()>

Commits the transaction that is currently active on this session.

This method may return an error with a crate::error::UNKNOWN_TRANSACTION_COMMIT_RESULT label. This label indicates that it is unknown whether the commit has satisfied the write concern associated with the transaction. If an error with this label is returned, it is safe to retry the commit until the write concern is satisfied or an error without the label is returned.

session.start_transaction(None).await?;
let result = coll.insert_one_with_session(doc! { "x": 1 }, None, &mut session).await?;
session.commit_transaction().await?;

This operation will retry once upon failure if the connection and encountered error support retryability. See the documentation here for more information on retryable writes.

pub async fn abort_transaction(&mut self) -> Result<()>

Aborts the transaction that is currently active on this session. Any open transaction will be aborted automatically in the Drop implementation of ClientSession.

session.start_transaction(None).await?;
match execute_transaction(&coll, &mut session).await {
    Ok(_) => session.commit_transaction().await?,
    Err(_) => session.abort_transaction().await?,
}

async fn execute_transaction(coll: &Collection<Document>, session: &mut ClientSession) -> Result<()> {
    coll.insert_one_with_session(doc! { "x": 1 }, None, session).await?;
    coll.delete_one_with_session(doc! { "y": 2 }, None, session).await?;
    Ok(())
}

This operation will retry once upon failure if the connection and encountered error support retryability. See the documentation here for more information on retryable writes.

pub async fn with_transaction<R, C, F>( &mut self, context: C, callback: F, options: impl Into<Option<TransactionOptions>> ) -> Result<R>where F: for<'a> FnMut(&'a mut ClientSession, &'a mut C) -> Pin<Box<dyn Future<Output = Result<R>> + Send + 'a>>,

Starts a transaction, runs the given callback, and commits or aborts the transaction. Transient transaction errors will cause the callback or the commit to be retried; other errors will cause the transaction to be aborted and the error returned to the caller. If the callback needs to provide its own error information, the Error::custom method can accept an arbitrary payload that can be retrieved via Error::get_custom.

If a command inside the callback fails, it may cause the transaction on the server to be aborted. This situation is normally handled transparently by the driver. However, if the application does not return that error from the callback, the driver will not be able to determine whether the transaction was aborted or not. The driver will then retry the callback indefinitely. To avoid this situation, the application MUST NOT silently handle errors within the callback. If the application needs to handle errors within the callback, it MUST return them after doing so.

Because the callback can be repeatedly executed and because it returns a future, the rust closure borrowing rules for captured values can be overly restrictive. As a convenience, with_transaction accepts a context argument that will be passed to the callback along with the session:

let coll = client.database("mydb").collection::<Document>("mycoll");
let my_data = "my data".to_string();
// This works:
session.with_transaction(
    (&coll, &my_data),
    |session, (coll, my_data)| async move {
        coll.insert_one_with_session(doc! { "data": *my_data }, None, session).await
    }.boxed(),
    None,
).await?;
/* This will not compile with a "variable moved due to use in generator" error:
session.with_transaction(
    (),
    |session, _| async move {
        coll.insert_one_with_session(doc! { "data": my_data }, None, session).await
    }.boxed(),
    None,
).await?;
*/

Trait Implementations

impl Debug for ClientSession

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for ClientSession

fn drop(&mut self)

Executes the destructor for this type.