PostgreSQL / CockroachDB异步和同步事务重试

异步和同步Postgres协议事务重试。

为Rust中的PostgreSQL兼容数据库提供序列化事务重试机制。

错误类型要求

事务闭包必须返回一个实现了Into<postgres_tx_retry::Error<E>>的错误类型，并且对于序列化失败判断，您的错误类型应该实现（根据启用的功能）

• 来自<deadpool::managed::PoolError<tokio_postgres::Error>>
• 来自<tokio_postgres::Error>
• 来自<r2d2::Error>

一个基本的错误类型可能是

pub enum MyError {
    Pool(deadpool::managed::PoolError<tokio_postgres::Error>),
    Postgres(tokio_postgres::Error),
    R2d2(r2d2::Error),
    Message(&'static str),
}

impl Into<postgres_tx_retry::Error<Self>> for MyError {
    fn into(self) -> postgres_tx_retry::Error<Self> {
        match self {
            Self::Pool(e) => postgres_tx_retry::Error::Pool(e),
            Self::Postgres(e) => postgres_tx_retry::Error::Postgres(e),
            Self::R2d2(e) => postgres_tx_retry::Error::R2d2(e),
            _ => postgres_tx_retry::Error::Other(self),
        }
    }
}

impl<E> From<postgres_tx_retry::Error<E>> for MyError
where
    E: Into<Self>,
{
    fn from(src: postgres_tx_retry::Error<E>) -> Self {
        match src {
            postgres_tx_retry::Error::Pool(e) => Self::Pool(e),
            postgres_tx_retry::Error::Postgres(e) => Self::Postgres(e),
            postgres_tx_retry::Error::R2d2(e) => Self::R2d2(e),
            postgres_tx_retry::Error::Other(e) => e.into(),
        }
    }
}

异步示例

let values = Arc::new("London");
let population = tx(&pool, COCKROACH_SAVEPOINT, |tx| {
    let values = values.clone();

    Box::pin(async move {
        let name = values.as_ref();
        let row = tx
            .query_one(
                "SELECT population FROM city WHERE name = $1",
                &[&name],
            )
            .await?;

        Ok(row.get(0))
    })
})
.await?;

同步示例

let name = "London";
let population = tx_sync(&db, COCKROACH_SAVEPOINT, |tx| {
    let row = tx.query_one(
        "SELECT population FROM city WHERE name = $1",
        &[&name],
    )?;

    Ok(row.get(0))
})?;