事务与数据的一致性

// 错误设计：订单和订单项分两个集合，必须用事务保证一致性
const session = client.startSession();
session.startTransaction();
try {
    await db.t_orders.insertOne(
        { orderId: "ORD001", userId: "U001", total: 299 },
        { session }
    );
    await db.t_order_items.insertMany([
        { orderId: "ORD001", productId: "P001", quantity: 1, price: 199 },
        { orderId: "ORD001", productId: "P002", quantity: 1, price: 100 }
    ], { session });
    await session.commitTransaction();
} catch (err) {
    await session.abortTransaction();
    throw err;
} finally {
    await session.endSession();
}

// 正确设计：嵌入式文档，单文档操作天然原子，无需事务
await db.t_orders.insertOne({
    orderId: "ORD001",
    userId: "U001",
    total: 299,
    items: [
        { productId: "P001", quantity: 1, price: 199 },
        { productId: "P002", quantity: 1, price: 100 }
    ]
});

from pymongo import MongoClient
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern
import logging

logger = logging.getLogger(__name__)


def transfer_funds(db, session, from_account, to_account, amount):
    """转账业务逻辑"""
    accounts = db.accounts

    # 扣款（CAS 校验余额）
    result = accounts.update_one(
        {"_id": from_account, "balance": {"$gte": amount}},
        {"$inc": {"balance": -amount}},
        session=session
    )
    if result.matched_count != 1:
        # 区分账户不存在与余额不足
        account = accounts.find_one({"_id": from_account}, session=session)
        if not account:
            raise Exception(f"扣款账户不存在: {from_account}")
        raise Exception(
            f"余额不足: {from_account}, 当前 {account['balance']}, 需扣 {amount}"
        )

    # 入账（必须校验目标账户存在）
    result = accounts.update_one(
        {"_id": to_account},
        {"$inc": {"balance": amount}},
        session=session
    )
    if result.matched_count != 1:
        raise Exception(f"入账账户不存在: {to_account}")


# 使用 with_transaction 回调 API（自动重试 TransientTransactionError）
client = MongoClient("mongodb://...")
db = client.bank

try:
    with client.start_session() as session:
        session.with_transaction(
            lambda s: transfer_funds(db, s, "A001", "A002", 100),
            read_concern=ReadConcern(level="snapshot"),
            write_concern=WriteConcern(w="majority", j=True)
        )
    logger.info("转账成功: A001 -> A002, 金额 100")
except Exception as e:
    logger.error(f"转账失败: {e}")
    raise

const { MongoClient } = require('mongodb');

async function transferFunds(client, fromAccount, toAccount, amount) {
    const session = client.startSession();

    try {
        // 使用 withTransaction 回调 API（自动重试 TransientTransactionError）
        await session.withTransaction(async () => {
            const accounts = client.db('bank').collection('accounts');

            // 扣款（CAS 校验余额）
            const debitResult = await accounts.updateOne(
                { _id: fromAccount, balance: { $gte: amount } },
                { $inc: { balance: -amount } },
                { session }
            );

            if (debitResult.matchedCount !== 1) {
                // 区分"账户不存在"与"余额不足"
                const account = await accounts.findOne(
                    { _id: fromAccount },
                    { session }
                );
                if (!account) {
                    throw new Error(`扣款账户不存在: ${fromAccount}`);
                }
                throw new Error(
                    `余额不足: ${fromAccount}, 当前 ${account.balance}, 需扣 ${amount}`
                );
            }

            // 入账（必须校验目标账户存在，否则钱会消失）
            const creditResult = await accounts.updateOne(
                { _id: toAccount },
                { $inc: { balance: amount } },
                { session }
            );

            if (creditResult.matchedCount !== 1) {
                throw new Error(`入账账户不存在: ${toAccount}`);
            }

        }, {
            readConcern: { level: 'snapshot' },
            writeConcern: { w: 'majority', j: true }
        });

        console.log(`转账成功: ${fromAccount} -> ${toAccount}, 金额 ${amount}`);
    } finally {
        await session.endSession();
    }
}

 // 错误：事务中包含耗时的外部调用
 session.startTransaction();
 await collection.updateOne({...}, {...}, { session });
 const result = await callExternalAPI();  // 外部调用，延迟不可控
 await collection.updateOne({...}, {...}, { session });
 session.commitTransaction();
 
 // 正确：外部调用放在事务外部
 const externalResult = await callExternalAPI();  // 先完成外部调用
 
 session.startTransaction();
 await collection.updateOne({...}, {...}, { session });
 await collection.updateOne({...}, {...}, { session });
 session.commitTransaction();

// 乐观锁实现
async function updateWithOptimisticLock(collection, id, updateFn, maxRetries = 5) {
    for (let i = 0; i < maxRetries; i++) {
        // 读取当前文档和版本号
        const doc = await collection.findOne({ _id: id });
        if (!doc) {
            throw new Error(`文档不存在: ${id}`);
        }
        const currentVersion = doc.version || 0;

        // 应用业务逻辑（业务异常直接外抛，不重试）
        const updates = updateFn(doc);

        // 带版本号条件更新
        const result = await collection.updateOne(
            { _id: id, version: currentVersion },
            {
                $set: updates,
                $inc: { version: 1 }
            }
        );

        // 用 matchedCount 判断 CAS 是否命中，而不是 modifiedCount
        if (result.matchedCount === 1) {
            return true;  // 更新成功
        }

        // 版本冲突，指数退避 + 随机抖动后重试
        const backoff = Math.random() * Math.pow(2, i) * 10;
        await new Promise(resolve => setTimeout(resolve, backoff));
    }
    throw new Error(`乐观锁更新失败，超过重试次数上限: ${maxRetries}`);
}

// 使用示例：库存扣减
await updateWithOptimisticLock(
    db.t_products,
    "P001",
    (product) => {
        if (product.stock < 1) throw new Error('库存不足');
        return { stock: product.stock - 1 };
    }
);

 // 默认：local（可能读到未持久化的数据）
 db.t_orders.find({ orderId: "ORD001" })
 
 // 推荐：majority（读取已持久化的数据）
 db.t_orders.find({ orderId: "ORD001" }).readConcern("majority")
 
 // 事务内：snapshot（快照隔离）
 session.startTransaction({
     readConcern: { level: "snapshot" },
     writeConcern: { w: "majority" }
 });
 
 // 连接串配置方式（全局生效）
 // mongodb://user:pwd@host1,host2,host3/db?readConcernLevel=majority

while (retries < maxRetries) {
    try {
        session.startTransaction();
        // ... 业务逻辑 ...
        session.commitTransaction();
        break;  // 成功退出
    } catch (error) {
        session.abortTransaction();
        if (error.hasErrorLabel("TransientTransactionError") && retries < maxRetries) {
            retries++;
            continue;  // 重试
        }
        throw error;  // 非瞬态错误，抛出
    }
}

// 1. 查看当前活跃事务
db.currentOp({ "transaction": { $exists: true } })

// 2. 检查事务执行时间
// 关注 transaction.timeActiveMicros 字段
// 超过 1 秒的事务需排查原因

// 3. 监控事务冲突
// 关注 serverStatus 中的 transactions 指标
db.serverStatus().transactions