SQLX 事务序列化冲突重试

示例说明

可串行化隔离级别把并发冲突的处理权交给应用：数据库回滚其中一个事务并返回 40001，正确响应是重放整个事务。识别码判定是关键——只重试冲突类错误，余额不足这类业务失败重试一万次也不会成功。重放必须覆盖事务的全部语句，begin 到 commit 是不可分的重放单元。

技术要点

• 40 开头错误码识别序列化冲突
• 行锁 SELECT FOR UPDATE 锁定读改写窗口
• 退避随尝试次数线性递增

代码解读

transfer 的循环最多四次尝试：冲突时按尝试数乘五十毫秒退避后重放；transfer_once 是完整事务体——先锁行读余额、校验、双更新、提交。is_serialization_failure 把数据库错误下转为 PgDatabaseError 后按码前缀判定。RowNotFound 复用为余额不足信号演示业务失败路径。

使用建议

热点账户冲突率高时改用队列化。事务体必须无外部副作用否则重放不安全。40 前缀包含死锁等类冲突，可按需收紧到 40001。

代码示例

use sqlx::postgres::PgDatabaseError;
use sqlx::{PgPool, Row};
use std::time::Duration;

async fn transfer(
    pool: &PgPool,
    from_id: i64,
    to_id: i64,
    amount: i64,
) -> Result<i64, sqlx::Error> {
    let mut attempts = 0;

    loop {
        attempts += 1;
        match transfer_once(pool, from_id, to_id, amount).await {
            Ok(balance) => return Ok(balance),
            Err(err) => {
                if is_serialization_failure(&err) && attempts < 4 {
                    let backoff = 50 * attempts;
                    tokio::time::sleep(Duration::from_millis(backoff)).await;
                    continue; // 序列化冲突：整块重放
                }
                return Err(err);
            }
        }
    }
}

fn is_serialization_failure(err: &sqlx::Error) -> bool {
    match err {
        sqlx::Error::Database(db) => db
            .downcast_ref::<PgDatabaseError>()
            .code()
            .starts_with("40"),
        _ => false,
    }
}

async fn transfer_once(
    pool: &PgPool,
    from_id: i64,
    to_id: i64,
    amount: i64,
) -> Result<i64, sqlx::Error> {
    let mut tx = pool.begin().await?;

    let balance: i64 = sqlx::query("SELECT balance FROM accounts WHERE id = $1 FOR UPDATE")
        .bind(from_id)
        .fetch_one(&mut *tx)
        .await?
        .get(0);

    if balance < amount {
        return Err(sqlx::Error::RowNotFound); // 余额不足：业务性失败不重试
    }

    sqlx::query("UPDATE accounts SET balance = balance - $1 WHERE id = $2")
        .bind(amount)
        .bind(from_id)
        .execute(&mut *tx)
        .await?;

    sqlx::query("UPDATE accounts SET balance = balance + $1 WHERE id = $2")
        .bind(amount)
        .bind(to_id)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(balance - amount)
}

#[tokio::main]
async fn main() -> Result<(), sqlx::Error> {
    let pool = PgPool::connect("postgres://user:pass@localhost/app").await?;
    let final_balance = transfer(&pool, 1, 2, 500).await?;
    println!("source balance now: {final_balance}");
    pool.close().await;
    Ok(())
}