MongoDB副本集下updateOne返回modifiedCount为0问题求助

环境配置 #

• MongoDB 4.2，3节点副本集部署
• Node.js 14.17.3
• mongodb包版本3.6.10
• 默认读关注：local，写关注：w:1

电商系统流程 #

1. 通过事务实现扣减库存与创建订单的原子性
2. 事务结束后调用Stripe发起支付
3. 支付成功后，使用updateOne更新已创建订单的支付ID

异常现象 #

流程正常运行，但高负载场景下偶现：

• 订单中缺失Stripe ID
• updateOne返回modifiedCount为0

简化代码 #

const allocate = async(sale, stocksToAllocate) => {
    const saleJSON = await this.mapping.encode(sale); // 将sale类实例转换为JSON对象
    
    const session = client.startSession();
    try {
        await session.withTransaction(async() => {
            if (sale._id)
                await this.repository.replaceOne({_id: sale._id}, saleJSON, {session});
            else
                await this.repository.insertOne(saleJSON, {session});
    
            for (const stockToAllocate of stocksToAllocate) {
                const stockJSON = await this.stockRepository.findOne({_id: stockToAllocate._id}, {session});
                const stock = await this.stockMapping.decode(stockJSON); // 将JSON对象转换为stock类实例
                if (stockToAllocate.count < 0) {
                    stock.replenish(-stockToAllocate.count, sale.delivery.shippingAt);
                } else if (stock.canAffect(stockToAllocate.count, sale.delivery.shippingAt)) {
                    stock.affect(stockToAllocate.count, sale.delivery.shippingAt);
                } else {
                    throw new ValidationError('库存不足', {id: stock._id});
                }
    
                const updatedStockJSON = await this.stockMapping.encode(stock);
                await this.stockRepository.replaceOne({_id: stockToAllocate._id}, updatedStockJSON, {session});
            }
        });
    } finally {
        await session.endSession();
    }

    const invoiceItem = {
        customer: sale.getStripeCustomerId(),
        subscription: sale.getStripeSubscriptionId(),
        amount: sale.getPrice().ttc,
        currency: 'EUR',
        description: sale.getName()
    };

    const updatedInvoiceItem = await stripe.invoiceItems.create(invoiceItem);
    await this.repository.updateOne({_id: sale._id}, {$set: {'payment.stripe.chargeId': updatedInvoiceItem.id}}); // 该updateOne有时返回modifiedCount为0
};

核心原因 #

1. 副本集读一致性缺失：默认读关注local允许从任意节点读取数据，事务提交后主节点的变更可能未同步到当前读操作命中的从节点，导致updateOne找不到最新提交的订单文档。
2. 事务提交后的同步延迟：高负载下副本集 oplog 同步速度变慢，后续updateOne若路由到滞后的从节点，会读取到未包含事务变更的旧数据，无法匹配目标订单。
3. 更新操作未绑定一致性上下文：updateOne未指定强一致性读关注或复用事务会话，无法保证读操作获取的是最新提交的数据。

解决方案 #

方案1：为updateOne指定majority读关注 #

强制更新操作从已同步事务变更的节点读取数据，确保能匹配到目标订单：

await this.repository.updateOne(
    {_id: sale._id}, 
    {$set: {'payment.stripe.chargeId': updatedInvoiceItem.id}},
    {readConcern: {level: 'majority'}}
);

方案2：复用事务会话执行更新 #

事务结束后继续使用同一个会话，会话会绑定到主节点（或已同步数据的节点），保证读一致性：

const allocate = async(sale, stocksToAllocate) => {
    const saleJSON = await this.mapping.encode(sale);
    
    const session = client.startSession();
    try {
        await session.withTransaction(async() => {
            // 原有事务逻辑
        });

        // 复用会话执行updateOne
        const invoiceItem = { /* ... */ };
        const updatedInvoiceItem = await stripe.invoiceItems.create(invoiceItem);
        await this.repository.updateOne(
            {_id: sale._id}, 
            {$set: {'payment.stripe.chargeId': updatedInvoiceItem.id}},
            {session} // 绑定会话
        );
    } finally {
        await session.endSession();
    }
};

方案3：全局调整默认读关注为majority #

如果业务对一致性要求较高，可修改客户端默认读关注，确保所有读操作都从已确认同步的节点获取数据：

const client = await MongoClient.connect(uri, {
    readConcern: { level: 'majority' },
    // 其他配置项
});

方案4：添加更新重试机制 #

针对modifiedCount为0的情况，增加指数退避重试，直到更新成功或达到重试上限：

const updateOrderWithRetry = async (repository, saleId, stripeChargeId, maxRetries = 3) => {
    let retries = 0;
    while (retries < maxRetries) {
        const result = await repository.updateOne(
            {_id: saleId}, 
            {$set: {'payment.stripe.chargeId': stripeChargeId}},
            {readConcern: {level: 'majority'}}
        );
        if (result.modifiedCount > 0) {
            return result;
        }
        retries++;
        await new Promise(resolve => setTimeout(resolve, 100 * retries));
    }
    throw new Error(`更新订单Stripe ID失败，重试${maxRetries}次后仍未成功`);
};

// 替换原updateOne调用
await updateOrderWithRetry(this.repository, sale._id, updatedInvoiceItem.id);

内容的提问来源于stack exchange，提问作者yves amsellem